Python Boto3로 S3에서 특정 날짜 이후 수정된 파일 목록 가져오는 방법

문제 상황

Python의 boto3 라이브러리를 사용하여 AWS S3 버킷에서 지정된 날짜·시간(타임스탬프) 이후에 수정된 파일만 골라 목록으로 가져오는 방법을 알아봅니다.

예시: S3의 Bucket_1/testfolder 경로에서 test.zip 파일이 2021-01-21 13:19:56.986445+00:00 이후에 수정되었다면 해당 파일을 결과 목록에 포함합니다.

해결 접근 방법 및 알고리즘

1단계 – 예외 처리를 위해 boto3와 botocore.exceptions(ClientError)를 임포트합니다.

2단계 – 함수 list_all_objects_based_on_last_modified는 s3_path와 last_modified_timestamp 두 개의 매개변수를 받습니다. last_modified_timestamp는 반드시 "2021-01-22 13:19:56.986445+00:00"과 같은 형식이어야 하며, boto3는 실행 환경의 지역과 무관하게 기본적으로 UTC 시간대를 기준으로 동작합니다.

3단계 – 입력받은 s3_path가 s3://bucket_name/key 형태의 올바른 AWS 경로인지 검증합니다.

4단계 – boto3 라이브러리로 AWS 세션(Session)을 생성합니다.

5단계 – S3용 AWS 리소스(resource) 객체를 생성합니다.

6단계 – list_objects 함수를 호출해 지정된 접두사(prefix) 아래의 모든 객체를 조회하고, 발생할 수 있는 예외를 처리합니다.

7단계 – 조회 결과는 딕셔너리 형태이며, 파일 단위 정보는 'Contents' 키에 담겨 있습니다. 이 값을 추출해 반복 처리합니다.

8단계 – 각 객체 역시 파일의 상세 정보를 담은 딕셔너리입니다. 여기서 각 파일의 LastModified 값을 꺼내 지정한 날짜·시간과 비교합니다.

9단계 – LastModified 값이 지정한 타임스탬프보다 크거나 같으면 전체 파일 경로(S3 URI)를 결과 리스트에 저장하고, 그렇지 않으면 무시합니다.

10단계 – 최종적으로 지정된 시점 이후에 수정된 파일들의 전체 경로 목록을 반환합니다.

전체 예제 코드

아래 코드는 마지막 수정 날짜(타임스탬프)를 기준으로 AWS S3에서 파일 목록을 필터링해 가져옵니다.

import boto3
from botocore.exceptions import ClientError

def list_all_objects_based_on_last_modified(s3_files_path, last_modified_timestamp):
    # s3 경로 형식 검증
    if 's3://' not in s3_files_path:
        raise Exception('Given path is not a valid s3 path.')

    session = boto3.session.Session()
    s3_resource = session.resource('s3')

    # 버킷 이름과 폴더(접두사) 분리
    bucket_token = s3_files_path.split('/')
    bucket = bucket_token[2]
    folder_path = bucket_token[3:]
    prefix = ""
    for path in folder_path:
        prefix = prefix + path + '/'

    try:
        result = s3_resource.meta.client.list_objects(Bucket=bucket, Prefix=prefix)
    except ClientError as e:
        raise Exception("boto3 client error in list_all_objects_based_on_last_modified function: " + e.__str__())
    except Exception as e:
        raise Exception("Unexpected error in list_all_objects_based_on_last_modified function of s3 helper: " + e.__str__())

    filtered_file_names = []
    for obj in result['Contents']:
        if str(obj["LastModified"]) >= str(last_modified_timestamp):
            full_s3_file = "s3://" + bucket + "/" + obj["Key"]
            filtered_file_names.append(full_s3_file)

    return filtered_file_names

# test.zip을 가져올 수 있는 타임스탬프 지정
print(list_all_objects_based_on_last_modified("s3://Bucket_1/testfolder", "2021-01-21 13:19:56.986445+00:00"))

# 이 시각 이후 수정된 파일이 없는 타임스탬프 지정
print(list_all_objects_based_on_last_modified("s3://Bucket_1/testfolder", "2021-01-21 13:19:56.986445+00:00"))

실행 결과

# test.zip을 가져올 수 있는 타임스탬프 지정
['s3://Bucket_1/testfolder/test.zip']

# 이 시각 이후 수정된 파일이 없는 타임스탬프 지정
[]

추가 팁

페이지네이션 주의: list_objects는 한 번의 호출로 최대 1,000개의 객체만 반환합니다. 객체 수가 많은 버킷이라면 list_objects_v2와 paginator(get_paginator('list_objects_v2'))를 사용해 전체 객체를 순회하는 것이 좋습니다.
시간대 비교: S3가 반환하는 LastModified는 UTC 기준 datetime 객체입니다. 문자열 비교보다는 두 값을 datetime으로 변환해 비교하면 더 안전하고 정확합니다.
권한 요구 사항: 대상 버킷에 대한 s3:ListBucket 권한이 있는 IAM 자격 증명이 필요하며, 필요하다면 AWS CLI의 aws configure 또는 환경 변수로 자격 증명을 미리 설정해 두어야 합니다.