はじめに
boto3でAWSリソースにアクセスするサンプルコードの覚書きです。
随時更新予定
S3
S3のサンプルコード集
KeyObjectの全取得
指定バケットのすべてのObjectをListで取得します。
取得する方法としてresourceとclientの2通りあります。resourceのほうが記述がシンプルで処理が高速とのこと。
resource
バケット内のファイルオブジェクトをListで取得します。
取得するオブジェクトにはフォルダオブジェクトが含まれるため末尾が/であることを判定して除外しています。
python：sample.py
from typing import List
import boto3
s3_resource = boto3.resource("s3")
def get_bucket_keys(bucket: str, key_prefix: str = "", **kwargs) -> List[str]:
params = {"Prefix": key_prefix}
params.update(kwargs)
objects = s3_resource.Bucket(bucket).objects.filter(**params)
return [o.key for o in objects if not o.key.endswith("/")]
詳細はドキュメント参照：
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Bucket.objects
client
list_objects_v2は一度に最大1000件まで取得するので、バケット内のObjectをすべて取得するにはTokenで制御する必要があります。
python：sample.py
from typing import List
import boto3
s3_client= boto3.client("s3")
def get_s3_objects(bucket: str, key_prefix: str = "", **kwargs) -> List[dict]:
result: list = []
params: dict = {"Bucket": bucket, "Prefix": key_prefix}
params.update(kwargs)
while True:
response = s3_client.list_objects_v2(**params)
if "Contents" in response:
result.extend(response["Contents"])
if "NextContinuationToken" in response:
params["ContinuationToken"] = response["NextContinuationToken"]
else:
break
return result
Contentsには、ObjectKey名のほかSizeなどが含まれます。
Contentsの内容は公式ドキュメント参照：
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.list_objects_v2
KeyObjectの全取得(応用)
clientで得た結果をファイル名とファイルサイズに整形するサンプル。
結果にはフォルダタイプのオブジェクトが含まれているので除外し、ファイルサイズは人読みやすいようにマネジメントコンソール画面での表記(ex: 12.3 KB)でフォーマットします。
python：sample.py
def get_s3_files_size(bucket: str, key_prefix: str = "", **kwargs) -> dict:
objects = get_s3_objects(bucket, key_prefix, **kwargs)
# Remove objects end with "/", means not file.
return {
o["Key"]: set_number_unit(o["Size"])
for o in objects
if not o["Key"].endswith("/")
}
def set_number_unit(num: int) -> str:
units = {
"TB": 1024 ** 4,
"GB": 1024 ** 3,
"MB": 1024 ** 2,
"KB": 1024,
"B": 1,
}
for unit, size in units.items():
if num >= size:
return f"{(num/size):.1f} {unit}"
else:
return f"{num} B"