【AWS】Boto3のよく使うメソッドのチートシート(S3編)【覚え書き】 #Python

はじめに

AWSのLambdaやGlueでコードを書くときによくBoto3というライブラリを使用します。
Boto3には多くのメソッドがありますがその中で個人的に比較的使用頻度の高いメソッドとその利用例のコードをこの記事でまとめました。
※今後必要に応じてアップデート予定です
※ここではS3の"オブジェクト"を"ファイル"と表現しています
この記事で紹介しているメソッド一覧

メソッド名	利用シーン
head_bucket	バケットの存在確認
head_object	ファイルの存在確認、アップロード日時などの情報の取得
list_objects_v2	フォルダ内のファイル一覧取得
get_object	ファイルの取得
put_object	ファイルのアップロード
delete_object	ファイルの削除

S3のバケットやファイルの存在確認

バケットの存在確認

head_bucket

<コード例>
・特定のS3バケットを探して存在しなければエラーを返す

sample.py

import boto3

def lambda_handler(event, context):
    # 存在確認するS3バケット名
    target_bucket_name = 'your-target-bucket-name'
    # Boto3クライアントの作成
    s3_client = boto3.client('s3')
    try:
        response = s3_client.head_bucket(Bucket=target_bucket_name)
        print(f"S3バケット '{target_bucket_name}' は存在します。")
        return {
            'statusCode': 200,
            'body': 'Bucket exists'
        }
    except ClientError as e:
        if e.response['Error']['Code'] == '404':
            error_message = f"S3バケット '{target_bucket_name}' は存在しません。"
            print(error_message)
            return {
                'statusCode': 404,
                'body': error_message
            }
        else:
            # その他のエラーに対する処理
            error_message = f"S3バケット '{target_bucket_name}' の確認中にエラーが発生しました。エラー: {str(e)}"
            print(error_message)
            return {
                'statusCode': 500,
                'body': error_message
            }

ファイルの存在確認や関連情報の取得

head_object

<コード例>
・ファイルのアップロード日時を取得する

sample.py

response = s3_client.head_object(Bucket=bucket_name, Key=object_key)
last_modified_time = response['LastModified']

※LastModified 以外に取得できる情報の例

コード内に指定する文字列	取得できる情報
ContentLength	オブジェクトのサイズ（バイト単位）
ETag	オブジェクトのETag（ハッシュ）
ContentType	オブジェクトのコンテンツタイプ
StorageClass	オブジェクトのストレージクラス
VersionId	バージョン管理が有効な場合、オブジェクトのバージョンID

S3内のファイルへのアクセス

特定のフォルダ内のファイル一覧取得

list_objects_v2

<コード例>
・特定のフォルダ内のファイル一覧を取得する

sample.py

def list_files_in_folder(bucket_name, folder_prefix):
    s3_client = boto3.client('s3')

    # フォルダ内のオブジェクト一覧を取得
    response = s3_client.list_objects_v2(
        Bucket=bucket_name, # 一覧を取得するフォルダが含まれるバケット
        Prefix=folder_prefix # 一覧を取得するフォルダ
    )

    # ファイルのキーを表示
    files = [obj['Key'] for obj in response.get('Contents', [])]
    return files

ファイルの取得

get_object

<コード例>
・S3にあるCSVファイルを取得して特定の文字が含まれる行を抽出する

sample.py

source_bucket = 'your-source-bucket'
source_key = 'path/to/source/file.csv'
# CSVファイルの読み込み
response = s3_client.get_object(Bucket=source_bucket, Key=source_key)
csv_content = response['Body'].read().decode('utf-8')

# CSVデータをDataFrameに読み込み
df = pd.read_csv(io.StringIO(csv_content))

# 特定のパターンの行を抽出（ここでは例として 'pattern' が含まれる行を抽出）
extracted_df = df[df['column_name'].str.contains('pattern', case=False, na=False)]
extracted_csv_content = extracted_df.to_csv(index=False)

ファイルのアップロード

put_object

<コード例>
・前項のget_objectのコード例で編集したCSVファイルをS3にアップロードする

sample.py

destination_bucket = 'your-destination-bucket'
destination_key = 'path/to/destination/extracted_data.csv'
# 抽出結果を新しいCSVファイルとして保存
s3_client.put_object(Body=extracted_csv_content, Bucket=destination_bucket, Key=destination_key)

ファイルの削除

delete_object

<コード例>
・指定したバケットとオブジェクトキー示すファイルを削除する

sample.py

target_bucket_name = 'your-target-bucket-name'
target_object_key = 'path/to/your/object-to-delete.txt'
s3_client.delete_object(Bucket=target_bucket_name, Key=target_object_key)