PythonでBoto3ライブラリを使用して、AWSリソースを使用して最終更新日に基づいてS3からファイルのリストを取得するにはどうすればよいですか？

問題の説明 − Pythonでboto3ライブラリを使用して、S3からファイルのリストを取得します。これらのファイルは、指定された日付のタイムスタンプの後に変更されます。

例 − 2021-01-21 13：19：56.986445 + 00：00以降に変更された場合は、S3のBucket_1/testfolderからtest.zipをリストアップします。

この問題を解決するためのアプローチ/アルゴリズム

ステップ1 −例外を処理するためにboto3およびbotocore例外をインポートします。

ステップ2 − s3_path およびlast_modified_timestamp 関数list_all_objects_based_on_last_modifiedの2つのパラメーターです。 「last_modified_timestamp」 「2021-01-2213：19：56.986445 + 00：00」の形式である必要があります。デフォルトでは、boto3は地理的な場所に関係なくUTCタイムゾーンを理解します。

ステップ3 − s3_pathがAWS形式でs3：//bucket_name/keyとして渡されていることを検証します。

ステップ4 −boto3ライブラリを使用してAWSセッションを作成します。

ステップ5 −S3用のAWSリソースを作成します。

ステップ6 −関数list_objectsを使用して、指定されたプレフィックスのすべてのオブジェクトを一覧表示し、例外がある場合はそれを処理します。

ステップ7 −上記の関数の結果は辞書であり、「コンテンツ」という名前のキーにすべてのファイルレベルの情報が含まれています。次に、オブジェクト内のバケットレベルの詳細を抽出します。

ステップ8 −これで、オブジェクトはファイルのすべての詳細を含む辞書にもなります。次に、 LastModifiedをフェッチします 各ファイルの詳細を確認し、指定された日付のタイムスタンプと比較します。

ステップ9 − LastModifiedの場合 指定されたタイムスタンプよりも大きい場合は、完全なファイル名を保存してください。それ以外の場合は無視してください。

ステップ10 −指定された日付のタイムスタンプの後に変更されたファイルのリストを返します。

例

次のコードは、最終更新日のタイムスタンプに基づいてAWSS3からファイルのリストを取得します-

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

#give a timestamp to fetch test.zip
print(list_all_objects_based_on_last_modified("s3://Bucket_1/testfolder" , "2021-01-21 13:19:56.986445+00:00"))
#give a timestamp no file is modified after that
print(list_all_objects_based_on_last_modified("s3://Bucket_1/testfolder" , "2021-01-21 13:19:56.986445+00:00"))

出力

#give a timestamp to fetch test.zip
[s3://Bucket_1/testfolder/test.zip]
#give a timestamp no file is modified after that
[]