📦
boto3 で S3 に格納されているファイルの最終更新日を key にして取得する
はじめに
今回は、boto3 で、S3から対象ファイルの最終更新日をkeyにして、取得したいファイル一覧表示させるところまでです。
背景として、処理したいファイルのみをファイル取得し、後続処理のSQSメッセージ送信処理につなげたいからです。
前提条件
- AWS アカウントあり
- AWS IMA作成済
- AWS CLI 設定済
実装中リポジトリ
やりたいことイメージ
目次
- はじめに、処理したい S3 Bucket のファイル一覧を AWS CLI
$ aws s3 ls <bucket name>で確認する -
boto3で、最終更新日、LastModifiedを keyにして、指定したファイル一覧のみ取得する - 取得したファイル一覧をリストに格納し、結果を表示
手順
1.はじめに、処理したい S3 Bucket のファイル一覧を $ aws s3 ls <bucket name> で確認する
このファイルの中から、2021-11-06 14:00:00 以降の時間でファイルを取得してみます。
aws CLI
$ aws s3 ls s3://bucketid001
2021-11-06 13:36:16 7613 data_211021.json
2021-11-06 13:36:17 7613 data_211022.json
2021-11-06 13:36:16 7613 data_211023.json
2021-11-06 13:36:16 7613 data_211024.json
2021-11-06 13:36:16 7530 data_211025.json
2021-11-06 18:08:57 115 response.json
2021-11-06 18:08:57 115 response2.json
2021-11-06 18:08:57 115 response3.json
2021-11-06 18:08:57 0 test.sql
2021-11-06 18:08:57 264 user.json
2021-11-06 18:08:57 3206 user.sql
2.boto3 で、最終更新日、LastModified を keyに して、指定したファイル一覧のみ取得する
- ファイルは、 2021-11-06 14:00:00 以降で取得する
model で ロジックを実装
api/models/storage.py
# model側
import datetime
import boto3
class Storage(object):
"""
AWS S3 Bucket 操作する model
"""
def __init__(self, client: str, bucket_name: str, region: str):
""" initialization parameter
params
------
client(str): s3
bucket_name(str): choice bucket name
region(str): Region code
"""
self.client = boto3.client(client)
self.resource_bucket = boto3.resource(client)
self.bucket_name = bucket_name
self.region = region
# 略 ...
def get_timestamp_file1(self) -> dict:
"""
Retrieve all the data in the specified bucket and return it in a dictionary with the value of last modified.
return
------
dict: {'file name': 'LastModified'}
ex) {'data_211021.json': '2021-11-06 09:35:53'}
"""
file_last_modified_dict = {}
my_bucket = self.resource_bucket.Bucket(self.bucket_name)
for obj in my_bucket.objects.all():
# 日本時間に調整するために、parameter に `hours=9` を指定
obj_last_modified = str(obj.get()['LastModified'] + datetime.timedelta(hours=9))[0:-6]
file_last_modified_dict[obj.key] = obj_last_modified
return file_last_modified_dict
def get_files_up_specified_timestamp(self, file_dict, filter_time='2021-11-06 14:00:00') -> list:
""" Get files up to the specified timestamp.
params
------
file_dict(dict): key: file_name, value: Last modified
filter_time(str): The time of the timestamp you want to retrieve.
return
------
List of retrieved files
"""
print(f"*** since: {filter_time} ***")
filter_files = []
for file_name, timestamp in file_dict.items():
if timestamp >= filter_time:
print(f"file: {file_name:20} timestamp: {timestamp}")
filter_files.append(file_name)
print('-' * 20)
return filter_files
controller でmodel で書いたロジックを呼びだす。
api/controllers/storage_controller.py
# controller で処理
import os
from dotenv import load_dotenv
from api.models import storage
from config import const
def s3_storage_management():
""" aws s3 management """
# call .env
load_dotenv()
region = os.environ.get('AWS_REGION')
management_storage = storage.Storage(client=const.CLIENT, # s3
bucket_name=const.BUCKET_NAME, # bucketid001
region=region) # 指定リージョン
# 関数呼び出し
file_dict = management_storage.get_timestamp_file1()
management_storage.get_files_up_specified_timestamp(file_dict)
main.pyで実行
main.py
# 実行ファイル
import api.controllers.storage_controller
if __name__ == '__main__':
api.controllers.storage_controller.s3_storage_management()
3.ファイル取得の結果確認
指定した、2021-11-06 14:00:00 以降のファイル一覧を取得することができました!
# OutPut
file: response.json timestamp: 2021-11-06 18:08:57
file: response2.json timestamp: 2021-11-06 18:08:57
file: response3.json timestamp: 2021-11-06 18:08:57
file: test.sql timestamp: 2021-11-06 18:08:57
file: user.json timestamp: 2021-11-06 18:08:57
file: user.sql timestamp: 2021-11-06 18:08:57
まとめ
boto3 で、格納ファイルの最終更新日(LastModified) を key にして取得することができました!
次回は、取得したファイル一覧で、SQSメッセージを作成したと思います。
ありがとうございました!!
参考
Discussion