Computer >> 컴퓨터 >  >> 프로그래밍 >> Python

Boto3로 AWS S3 버킷의 모든 객체를 페이지네이션하는 방법

문제 상황: Python의 boto3 라이브러리를 사용하여 계정에 생성된 AWS Glue Data Catalog의 S3 버킷에 있는 모든 객체를 페이지네이션(pagination) 방식으로 조회하는 방법을 알아봅니다.

문제 해결 접근 방식 및 알고리즘

  • 1단계: 예외 처리를 위해 boto3와 botocore의 예외 클래스를 임포트합니다.

  • 2단계: max_items, page_size, starting_token은 선택적(optional) 파라미터이며, bucket_name만 필수(required) 파라미터입니다.

    • max_items: 반환할 총 레코드 수를 의미합니다. 사용 가능한 레코드 수가 max_items보다 많으면, 응답에 NextToken이 포함되어 이후 페이지네이션을 이어갈 수 있습니다.

    • page_size: 한 페이지당 포함될 레코드 수(페이지 크기)를 의미합니다.

    • starting_token: 페이지네이션을 재개할 때 사용하며, 이전 응답에서 가져온 마지막 키(last key)를 기준으로 다음 데이터를 조회합니다.

  • 3단계: boto3 라이브러리를 사용해 AWS 세션(session)을 생성합니다. 기본 프로필(default profile)에 region_name이 설정되어 있지 않다면, 세션 생성 시 region_name을 명시적으로 전달해야 합니다.

  • 4단계: S3 서비스를 위한 AWS 클라이언트(client)를 생성합니다.

  • 5단계: list_objects를 사용하여 S3 버킷의 객체 정보를 담는 paginator 객체를 생성합니다.

  • 6단계: paginate 함수를 호출하고, max_items, page_size, starting_token은 PaginationConfig 파라미터로, bucket_name은 Bucket 파라미터로 전달합니다.

  • 7단계: max_size와 page_size 설정에 따라 해당 개수만큼의 레코드가 반환됩니다.

  • 8단계: 페이지네이션 과정에서 오류가 발생할 경우 일반 예외(generic exception)를 처리합니다.

예제 코드

아래 코드를 사용하면 계정에 생성된 S3 버킷의 모든 객체를 페이지네이션 방식으로 조회할 수 있습니다.

import boto3
from botocore.exceptions import ClientError

def paginate_through_all_objects_s3_bucket(bucket_name, max_items=None, page_size=None, starting_token=None):
    session = boto3.session.Session()
    s3_client = session.client('s3')
    try:
        paginator = s3_client.get_paginator('list_objects')
        response = paginator.paginate(
            Bucket=bucket_name,
            PaginationConfig={
                'MaxItems': max_items,
                'PageSize': page_size,
                'StartingToken': starting_token
            }
        )
        return response
    except ClientError as e:
        raise Exception("boto3 client error in paginate_through_all_objects_s3_bucket: " + str(e))
    except Exception as e:
        raise Exception("Unexpected error in paginate_through_all_objects_s3_bucket: " + str(e))

# 첫 번째 실행
a = paginate_through_all_objects_s3_bucket('s3-test-bucket', 2, 5)
print(*a)

# 두 번째 실행
for items in a:
    next_token = items['Contents'][max_items - 1]['Key']
    b = paginate_through_all_objects_s3_bucket('s3-test-bucket', 2, 5, next_token)
    print(*b)

실행 결과

첫 번째 실행에서는 버킷의 앞부분 객체들이 반환되며, IsTruncated 값이 True인 것을 통해 아직 조회하지 않은 객체가 더 남아 있음을 확인할 수 있습니다.

# 첫 번째 실행
{'ResponseMetadata': {'RequestId': '5AE5VF7RK', 'HostId': **************, 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amz-id-2': **************, 'x-amz-request-id': '5AE5VF7RK', 'date': 'Sat, 03 Apr 2021 07:33:28 GMT', 'x-amz-bucket-region': 'us-east-1', 'content-type': 'application/xml', 'transfer-encoding': 'chunked', 'server': 'AmazonS3'}, 'RetryAttempts': 0}, 'IsTruncated': True, 'Marker': '',
'Contents': [{'Key': 'analytics-s3i/param.json', 'LastModified': datetime.datetime(2020, 10, 29, 19, 50, 55, tzinfo=tzutc()), 'ETag': '"e6f6b8e02"', 'Size': 1554, 'StorageClass': 'STANDARD', 'Owner': {'DisplayName': 'AWS.Development', 'ID': '92************************0'}}, {'Key': 'analytics-s3i/params.json', 'LastModified': datetime.datetime(2021, 3, 10, 20, 10, 47, tzinfo=tzutc()), 'ETag': '"22a4bf70c1ed2612"', 'Size': 1756, 'StorageClass': 'STANDARD', 'Owner': {'DisplayName': 'AWS.Development', 'ID': '92************************************************'}}], 'Name': 's3-test-bucket', 'Prefix': '', 'MaxKeys': 5, 'EncodingType': 'url', 'CommonPrefixes': None}

두 번째 실행에서는 이전 응답의 마지막 키(analytics-s3i/params.json)를 starting_token으로 전달하여, 그 다음 위치부터 나머지 객체들을 이어서 조회합니다.

# 두 번째 실행
{'ResponseMetadata': {'RequestId': '5AEAWX', 'HostId': 'Uc********************************', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amz-id-2': *****************************, 'x-amz-request-id': '5AEAWX', 'date': 'Sat, 03 Apr 2021 07:33:28 GMT', 'x-amz-bucket-region': 'us-east-1', 'content-type': 'application/xml', 'transfer-encoding': 'chunked', 'server': 'AmazonS3'}, 'RetryAttempts': 0}, 'IsTruncated': True, 'Marker': 'analytics-s3i/params.json',
'Contents': [{'Key': 'analytics-s3i/template.json', 'LastModified': datetime.datetime(2020, 10, 29, 19, 50, 55, tzinfo=tzutc()), 'ETag': '"3af411f"', 'Size': 21334, 'StorageClass': 'STANDARD', 'Owner': {'DisplayName': 'AWS.Development', 'ID': '92************************'}}, {'Key': 'analytics-s3i2/param.json', 'LastModified': datetime.datetime(2020, 11, 20, 17, 32, 45, tzinfo=tzutc()), 'ETag': '"04c29cf86888f99"', 'Size': 1695, 'StorageClass': 'STANDARD', 'Owner': {'DisplayName': 'AWS.Development', 'ID': '92************************'}}], 'Name': 's3-test-bucket', 'Prefix': '', 'MaxKeys': 5, 'EncodingType': 'url', 'CommonPrefixes': None}

정리

Boto3의 paginator 기능을 활용하면 대량의 S3 객체도 max_items와 page_size를 조절해 효율적으로 나누어 조회할 수 있습니다. 또한 NextToken(마지막 키)을 활용하면 중단된 지점부터 안정적으로 페이지네이션을 재개할 수 있으므로, 대규모 데이터 처리나 배치 작업에 유용하게 활용할 수 있습니다.