AWS S3をPythonで操作する — boto3の基本をまとめた

はじめに

S3はAWSで一番使う頻度が高いサービスと言っても過言ではない。

FastAPIからファイルをアップロードする、データパイプラインでCSVを読み書きする、Snowflakeとの連携でデータを受け渡すなど、至るところでS3を触ることになった。都度調べていたのでboto3の操作を一度まとめておく。

---

インストールと設定

pip install boto3

認証の設定

# パターン① 環境変数（ローカル開発）
# AWS_ACCESS_KEY_ID と AWS_SECRET_ACCESS_KEY を設定

# パターン② IAMロール（EC2/ECS/Lambda）
# 設定不要。boto3が自動でロールから認証情報を取得

# パターン③ プロファイル（複数AWSアカウントを使う場合）
# ~/.aws/credentials にプロファイルを設定

import boto3

# 基本（IAMロールやデフォルトプロファイルを使う）
s3 = boto3.client('s3', region_name='ap-northeast-1')

# リソース形式（オブジェクト指向的なAPI）
s3_resource = boto3.resource('s3', region_name='ap-northeast-1')

# プロファイルを明示する場合
session = boto3.Session(profile_name='my-profile')
s3      = session.client('s3')

clientとresourceの2種類がある。clientは低レベルAPIで全操作が使える。resourceはオブジェクト指向的で書きやすいが一部操作はclientでしかできない。実務ではclientをメインに使っている。

---

バケット操作

import boto3

s3 = boto3.client('s3', region_name='ap-northeast-1')

# バケットの作成
s3.create_bucket(
    Bucket                    = 'my-new-bucket',
    CreateBucketConfiguration = {
        'LocationConstraint': 'ap-northeast-1'
    }
)

# バケットの一覧
response = s3.list_buckets()
for bucket in response['Buckets']:
    print(bucket['Name'], bucket['CreationDate'])

# バケットの削除（中身が空である必要がある）
s3.delete_bucket(Bucket='my-empty-bucket')

バケット名はAWS全体でユニークである必要がある。my-bucketのような名前はすでに使われていることが多い。my-app-2024-uploadsのように具体的な名前にする。

---

ファイルのアップロード

# ① ローカルファイルをアップロード
s3.upload_file(
    Filename = '/local/path/to/file.csv',
    Bucket   = 'my-bucket',
    Key      = 'data/2024/04/file.csv',
)

# ② バイトデータをアップロード
s3.put_object(
    Bucket      = 'my-bucket',
    Key         = 'reports/summary.json',
    Body        = b'{"total": 1000}',
    ContentType = 'application/json',
)

# ③ ファイルオブジェクトをアップロード
with open('/local/file.txt', 'rb') as f:
    s3.upload_fileobj(
        f,
        'my-bucket',
        'uploads/file.txt'
    )

# ④ 文字列をアップロード
content = "CSV内容\nrow1,row2"
s3.put_object(
    Bucket      = 'my-bucket',
    Key         = 'data/output.csv',
    Body        = content.encode('utf-8'),
    ContentType = 'text/csv; charset=utf-8',
)

メタデータとタグ

s3.upload_file(
    Filename  = '/local/file.csv',
    Bucket    = 'my-bucket',
    Key       = 'data/file.csv',
    ExtraArgs = {
        'ContentType': 'text/csv',
        'Metadata': {
            'uploaded-by': 'pipeline-v2',
            'source':      'orders-api',
        },
        'Tagging': 'env=production&project=myapp',
    }
)

---

ファイルのダウンロード

# ① ローカルファイルにダウンロード
s3.download_file(
    Bucket   = 'my-bucket',
    Key      = 'data/file.csv',
    Filename = '/local/downloaded.csv',
)

# ② メモリに読み込む
response = s3.get_object(
    Bucket = 'my-bucket',
    Key    = 'data/file.csv',
)
content = response['Body'].read()  # bytes
text    = content.decode('utf-8')

# ③ ファイルオブジェクトに書き込む
with open('/local/output.csv', 'wb') as f:
    s3.download_fileobj('my-bucket', 'data/file.csv', f)

DataFrameとして読み込む

import pandas as pd
import io

def read_csv_from_s3(bucket: str, key: str) -> pd.DataFrame:
    response = s3.get_object(Bucket=bucket, Key=key)
    return pd.read_csv(
        io.BytesIO(response['Body'].read()),
        encoding='utf-8',
    )

def read_parquet_from_s3(bucket: str, key: str) -> pd.DataFrame:
    response = s3.get_object(Bucket=bucket, Key=key)
    return pd.read_parquet(io.BytesIO(response['Body'].read()))

# 使う側
df = read_csv_from_s3('my-bucket', 'data/orders.csv')
print(df.head())

---

DataFrameをS3に書き込む

import pandas as pd
import io

def write_csv_to_s3(
    df:     pd.DataFrame,
    bucket: str,
    key:    str,
    index:  bool = False,
) -> None:
    buffer = io.StringIO()
    df.to_csv(buffer, index=index, encoding='utf-8')
    s3.put_object(
        Bucket      = bucket,
        Key         = key,
        Body        = buffer.getvalue().encode('utf-8'),
        ContentType = 'text/csv; charset=utf-8',
    )
    print(f"書き込み完了: s3://{bucket}/{key}")

def write_parquet_to_s3(
    df:     pd.DataFrame,
    bucket: str,
    key:    str,
) -> None:
    buffer = io.BytesIO()
    df.to_parquet(buffer, index=False, engine='pyarrow')
    buffer.seek(0)
    s3.put_object(
        Bucket      = bucket,
        Key         = key,
        Body        = buffer.getvalue(),
        ContentType = 'application/octet-stream',
    )
    print(f"書き込み完了: s3://{bucket}/{key}")

# 使う側
df = pd.DataFrame({'name': ['田中', '鈴木'], 'age': [28, 35]})
write_csv_to_s3(df, 'my-bucket', 'output/users.csv')
write_parquet_to_s3(df, 'my-bucket', 'output/users.parquet')

---

ファイル一覧の取得

# 基本的な一覧取得
response = s3.list_objects_v2(
    Bucket = 'my-bucket',
    Prefix = 'data/2024/04/',
)

if 'Contents' in response:
    for obj in response['Contents']:
        print(obj['Key'], obj['Size'], obj['LastModified'])

ページネーション対応（1000件以上）

def list_all_files(bucket: str, prefix: str = '') -> list[dict]:
    """1000件を超えるファイルも全件取得する"""
    paginator = s3.get_paginator('list_objects_v2')
    pages     = paginator.paginate(Bucket=bucket, Prefix=prefix)

    files = []
    for page in pages:
        if 'Contents' in page:
            files.extend(page['Contents'])
    return files

# 使う側
files = list_all_files('my-bucket', 'data/2024/')
print(f"合計: {len(files)}件")
for f in files:
    print(f['Key'], f['Size'])

list_objects_v2は1回のリクエストで最大1000件しか返さない。paginatorを使うと1000件を超えるファイルも全件取得できる。これを知らずに最初は1001件目以降が取れなくて詰まった。

特定の拡張子のみ取得

def list_csv_files(bucket: str, prefix: str) -> list[str]:
    """CSVファイルのKeyのみ返す"""
    files = list_all_files(bucket, prefix)
    return [f['Key'] for f in files if f['Key'].endswith('.csv')]

csv_keys = list_csv_files('my-bucket', 'data/2024/')
print(csv_keys)

---

ファイルの削除

# 1ファイル削除
s3.delete_object(
    Bucket = 'my-bucket',
    Key    = 'data/old-file.csv',
)

# 複数ファイルを一括削除（最大1000件）
s3.delete_objects(
    Bucket = 'my-bucket',
    Delete = {
        'Objects': [
            {'Key': 'data/file1.csv'},
            {'Key': 'data/file2.csv'},
            {'Key': 'data/file3.csv'},
        ],
        'Quiet': True,  # エラーのみレスポンスに含める
    }
)

# プレフィックス以下を全削除
def delete_all_objects(bucket: str, prefix: str) -> int:
    files = list_all_files(bucket, prefix)
    if not files:
        return 0

    # 1000件ずつ削除
    deleted = 0
    for i in range(0, len(files), 1000):
        batch = files[i:i + 1000]
        s3.delete_objects(
            Bucket = bucket,
            Delete = {
                'Objects': [{'Key': f['Key']} for f in batch],
                'Quiet':   True,
            }
        )
        deleted += len(batch)

    print(f"{deleted}件削除しました")
    return deleted

---

署名付きURL

# ダウンロード用の署名付きURL（一時的なアクセスURL）
url = s3.generate_presigned_url(
    'get_object',
    Params    = {
        'Bucket': 'my-bucket',
        'Key':    'private/report.pdf',
    },
    ExpiresIn = 3600,  # 1時間有効
)
print(url)
# https://my-bucket.s3.ap-northeast-1.amazonaws.com/private/report.pdf?...

# アップロード用の署名付きURL（クライアントから直接アップロード）
presigned_post = s3.generate_presigned_post(
    Bucket     = 'my-bucket',
    Key        = 'uploads/${filename}',
    Fields     = {'Content-Type': 'image/jpeg'},
    Conditions = [
        ['content-length-range', 1, 10 * 1024 * 1024],  # 10MB以下
        ['eq', '$Content-Type', 'image/jpeg'],
    ],
    ExpiresIn  = 300,  # 5分有効
)

FastAPIでの署名付きURL活用

from fastapi import FastAPI, HTTPException
import boto3

app = FastAPI()
s3  = boto3.client('s3', region_name='ap-northeast-1')

@app.get("/files/{file_key:path}/download-url")
def get_download_url(file_key: str):
    """プライベートファイルの一時ダウンロードURLを発行"""
    try:
        # ファイルの存在確認
        s3.head_object(Bucket='my-bucket', Key=file_key)
    except s3.exceptions.ClientError as e:
        if e.response['Error']['Code'] == '404':
            raise HTTPException(status_code=404, detail="ファイルが見つかりません")
        raise

    url = s3.generate_presigned_url(
        'get_object',
        Params    = {'Bucket': 'my-bucket', 'Key': file_key},
        ExpiresIn = 3600,
    )
    return {'url': url, 'expires_in': 3600}

@app.post("/files/upload-url")
def get_upload_url(filename: str, content_type: str):
    """クライアントが直接S3にアップロードするためのURLを発行"""
    import uuid
    key = f"uploads/{uuid.uuid4()}/{filename}"

    presigned = s3.generate_presigned_post(
        Bucket     = 'my-bucket',
        Key        = key,
        Fields     = {'Content-Type': content_type},
        Conditions = [
            ['content-length-range', 1, 50 * 1024 * 1024],  # 50MB以下
            ['eq', '$Content-Type', content_type],
        ],
        ExpiresIn  = 300,
    )
    return {'upload_url': presigned['url'], 'fields': presigned['fields'], 'key': key}

---

ファイルのコピーと移動

# バケット内でコピー
s3.copy_object(
    CopySource = {'Bucket': 'my-bucket', 'Key': 'source/file.csv'},
    Bucket     = 'my-bucket',
    Key        = 'destination/file.csv',
)

# 別バケットにコピー
s3.copy_object(
    CopySource = {'Bucket': 'source-bucket', 'Key': 'file.csv'},
    Bucket     = 'dest-bucket',
    Key        = 'file.csv',
)

# 移動（コピー後に元を削除）
def move_object(
    bucket:     str,
    source_key: str,
    dest_key:   str,
) -> None:
    s3.copy_object(
        CopySource = {'Bucket': bucket, 'Key': source_key},
        Bucket     = bucket,
        Key        = dest_key,
    )
    s3.delete_object(Bucket=bucket, Key=source_key)
    print(f"移動完了: {source_key} → {dest_key}")

---

大きなファイルのマルチパートアップロード

5GBを超えるファイルはマルチパートアップロードが必要。

import boto3
from boto3.s3.transfer import TransferConfig

# TransferConfigでしきい値を設定
config = TransferConfig(
    multipart_threshold  = 1024 * 25,    # 25MBを超えたらマルチパート
    max_concurrency      = 10,           # 並列アップロード数
    multipart_chunksize  = 1024 * 25,    # チャンクサイズ
    use_threads          = True,
)

s3 = boto3.client('s3', region_name='ap-northeast-1')

# 通常のupload_fileで自動的にマルチパートになる
s3.upload_file(
    '/local/huge-file.csv',
    'my-bucket',
    'data/huge-file.csv',
    Config   = config,
    Callback = lambda bytes_transferred: print(
        f"転送済み: {bytes_transferred / 1024 / 1024:.1f}MB"
    ),
)

---

エラーハンドリング

from botocore.exceptions import ClientError, NoCredentialsError

def safe_download(bucket: str, key: str, local_path: str) -> bool:
    try:
        s3.download_file(bucket, key, local_path)
        return True

    except ClientError as e:
        error_code = e.response['Error']['Code']

        if error_code == '404' or error_code == 'NoSuchKey':
            print(f"ファイルが見つかりません: s3://{bucket}/{key}")
        elif error_code == '403':
            print(f"アクセス権限がありません: s3://{bucket}/{key}")
        elif error_code == 'NoSuchBucket':
            print(f"バケットが存在しません: {bucket}")
        else:
            print(f"S3エラー ({error_code}): {e}")
        return False

    except NoCredentialsError:
        print("AWS認証情報が設定されていません")
        return False

# ファイルの存在確認（例外を使わずに確認）
def object_exists(bucket: str, key: str) -> bool:
    try:
        s3.head_object(Bucket=bucket, Key=key)
        return True
    except ClientError as e:
        if e.response['Error']['Code'] in ('404', 'NoSuchKey'):
            return False
        raise

---

S3イベント通知（Lambdaとの連携）

# S3にファイルがアップロードされたときにLambdaを起動する場合
# Lambda側でイベントを受け取る

def lambda_handler(event, context):
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key    = record['s3']['object']['key']
        size   = record['s3']['object']['size']

        print(f"新しいファイル: s3://{bucket}/{key} ({size}バイト)")

        # ファイルを処理する
        response = s3.get_object(Bucket=bucket, Key=key)
        content  = response['Body'].read().decode('utf-8')

        # CSVとして処理
        import io, pandas as pd
        df = pd.read_csv(io.StringIO(content))
        print(f"取得件数: {len(df)}")

S3のイベント通知 + Lambdaで「ファイルがアップロードされたら自動処理する」というパイプラインが作れる。Snowflakeのパイプラインで使っているパターン。

---

よく使う操作まとめ

# よく使うパターンをまとめた関数集

class S3Helper:
    def __init__(self, bucket: str, region: str = 'ap-northeast-1'):
        self.bucket = bucket
        self.client = boto3.client('s3', region_name=region)

    def exists(self, key: str) -> bool:
        try:
            self.client.head_object(Bucket=self.bucket, Key=key)
            return True
        except ClientError:
            return False

    def read_text(self, key: str, encoding: str = 'utf-8') -> str:
        response = self.client.get_object(Bucket=self.bucket, Key=key)
        return response['Body'].read().decode(encoding)

    def write_text(self, key: str, content: str, encoding: str = 'utf-8') -> None:
        self.client.put_object(
            Bucket = self.bucket,
            Key    = key,
            Body   = content.encode(encoding),
        )

    def read_json(self, key: str) -> dict:
        import json
        return json.loads(self.read_text(key))

    def write_json(self, key: str, data: dict, indent: int = 2) -> None:
        import json
        self.write_text(key, json.dumps(data, ensure_ascii=False, indent=indent))

    def read_df(self, key: str, **kwargs) -> pd.DataFrame:
        response = self.client.get_object(Bucket=self.bucket, Key=key)
        if key.endswith('.parquet'):
            return pd.read_parquet(io.BytesIO(response['Body'].read()))
        return pd.read_csv(io.BytesIO(response['Body'].read()), **kwargs)

    def write_df(self, key: str, df: pd.DataFrame, index: bool = False) -> None:
        if key.endswith('.parquet'):
            write_parquet_to_s3(df, self.bucket, key)
        else:
            write_csv_to_s3(df, self.bucket, key, index=index)

    def presigned_url(self, key: str, expires: int = 3600) -> str:
        return self.client.generate_presigned_url(
            'get_object',
            Params    = {'Bucket': self.bucket, 'Key': key},
            ExpiresIn = expires,
        )

# 使う側
s3 = S3Helper('my-bucket')

if s3.exists('data/file.csv'):
    df = s3.read_df('data/file.csv')
    df['processed'] = True
    s3.write_df('output/processed.csv', df)
    url = s3.presigned_url('output/processed.csv')
    print(f"ダウンロードURL: {url}")

---

まとめ

操作	メソッド
アップロード（ファイル）	upload_file()
アップロード（バイト）	put_object()
ダウンロード（ファイル）	download_file()
ダウンロード（メモリ）	get_object()
一覧取得	list_objects_v2() + paginator
削除	delete_object() / delete_objects()
コピー	copy_object()
存在確認	head_object()
署名付きURL	generate_presigned_url()

• 1000件を超えるファイル一覧はpaginatorを使う
• DataFrameの読み書きはio.BytesIO/StringIOを介してメモリ上でやり取りする
• プライベートファイルのアクセスは署名付きURLで制御する
• 認証はIAMロールを使ってアクセスキーをコードに書かない