PythonのBoto3ライブラリを使って、最終更新日に基づいてS3からファイル一覧を取得する方法

問題概要

Pythonのboto3ライブラリを使用して、指定した日時（タイムスタンプ）以降に変更されたS3上のファイル一覧を取得します。

例： Bucket_1/testfolder 内の test.zip が「2021-01-21 13:19:56.986445+00:00」以降に変更されている場合、そのファイルを出力対象とします。

解決のためのアプローチとアルゴリズム

ステップ1

例外処理のために boto3 と botocore.exceptions をインポートします。

ステップ2

関数 list_all_objects_based_on_last_modified のパラメータとして s3_path と last_modified_timestamp の2つを受け取ります。「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セッションを作成します。

ステップ5

S3用のAWSリソースを作成します。

ステップ6

list_objects 関数を使って、指定したプレフィックス配下のすべてのオブジェクトを取得し、発生した例外を適切にハンドリングします。

ステップ7

上記関数の戻り値は辞書型で、「Contents」というキーにすべてのファイル単位の情報が格納されています。ここからバケット単位の詳細情報を抽出します。

ステップ8

各オブジェクトもまた辞書型で、ファイルに関するすべての詳細情報を保持しています。そこから各ファイルの LastModified（最終更新日時）を取得し、指定された日時と比較します。

ステップ9

LastModified が指定したタイムスタンプより大きい場合は完全なファイル名を保存し、そうでない場合は無視します。

ステップ10

指定した日時以降に変更されたファイルのリストを返します。

サンプルコード

以下のコードは、最終更新日のタイムスタンプに基づいてAWS S3からファイル一覧を取得します。

import boto3
from botocore.exceptions import ClientError

def list_all_objects_based_on_last_modified(s3_files_path,
last_modified_timestamp):
    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]
#その日時以降に変更されたファイルが存在しないタイムスタンプを指定
[]