boto3でS3を操作するには、boto3.client("s3") を作り、アップロードは upload_file、ダウンロードは download_file、一覧は list_objects_v2 を呼びます。 ファイルを介さずメモリ上のデータを扱うなら put_object と get_object、存在確認は head_object、まとめて消すなら delete_objects、という対応です。

この記事では、やりたいことからメソッドを引ける対応表を先頭に置き、そのまま貼って動くコードと実行結果を順に載せます。1回の呼び出しで返るのは1000件までという一覧取得の決まりと paginator、Prefix と Delimiter による「フォルダ」の取り出し、キーやバケットが無いとき・権限が無いときに届く例外をエラーコード別に捕まえる書き方まで、手元で動かした出力をもとに整理しました。

確認環境:Ubuntu 24.04 / Python 3.11.15 / boto3 1.43.103(botocore 1.43.103)。AWS の API をローカルで再現するツール(moto 5.2.3)で実行しています。掲載コードは接続先の指定を外してあり、実際の AWS でそのまま使える形です。

boto3のS3操作 早見表(やりたいこと別)

やりたいこと client のメソッド 最小の書き方 戻り値・ポイント
ローカルのファイルを上げる upload_file s3.upload_file("a.csv", BUCKET, "dir/a.csv") 戻り値は None。大きなファイルは自動で分割送信
メモリ上のデータを上げる put_object s3.put_object(Bucket=BUCKET, Key=k, Body=b"...") ETag を含む dict。Body は bytes か str かファイル
ファイルとして落とす download_file s3.download_file(BUCKET, "dir/a.csv", "a.csv") 戻り値は None。ローカルに保存
中身をメモリに読む get_object s3.get_object(Bucket=BUCKET, Key=k)["Body"].read() Body はストリーム。read() で bytes
一覧を取る list_objects_v2 s3.list_objects_v2(Bucket=BUCKET, Prefix="dir/") 1回で最大1000件。0件のとき Contents キーが無い
1000件を超える一覧 get_paginator s3.get_paginator("list_objects_v2").paginate(...) ページを順に返す。トークン管理は不要
「フォルダ」の一覧 list_objects_v2 Prefix="dir/", Delimiter="/" CommonPrefixes に下位フォルダが入る
存在確認・サイズや種類の取得 head_object s3.head_object(Bucket=BUCKET, Key=k) 本文を取らない。無いときはコード 404
1件消す delete_object s3.delete_object(Bucket=BUCKET, Key=k) 無いキーでも 204 で成功扱い
まとめて消す delete_objects Delete={"Objects": [{"Key": k}, ...]} 1回で1000件まで
コピー・移動 copy s3.copy({"Bucket": B, "Key": src}, B, dst) 移動はコピーのあと delete_object
期限つきの URL を発行 generate_presigned_url s3.generate_presigned_url("get_object", Params=..., ExpiresIn=600) 通信せずに URL 文字列を返す

本記事の内容は、ご自由にお使いください。
ご利用の際は、出典として本ページへのリンクを記載いただけますようお願いします。

(記載例)出典:株式会社RJC「boto3でS3を操作する方法|アップロード・ダウンロード・一覧取得」

準備:boto3の導入とclientの作成

boto3 は pip で入れます。認証情報は aws configure で作った設定、環境変数、実行ロールのいずれかから自動で読み込まれるので、コードにキーを書く必要はありません。

pip install boto3
import boto3

s3 = boto3.client("s3")          # リージョンは設定や環境変数 AWS_DEFAULT_REGION から決まる
BUCKET = "rjc-sample-bucket"

s3.create_bucket(
    Bucket=BUCKET,
    CreateBucketConfiguration={"LocationConstraint": "ap-northeast-1"},
)
print([b["Name"] for b in s3.list_buckets()["Buckets"]])

実行結果:

['rjc-sample-bucket']

以降のコードは、この s3 と BUCKET をそのまま使います。S3 にフォルダという実体はなく、reports/2026/report.csv のようなスラッシュ入りの文字列全体が1つの「キー」です。 この前提を押さえておくと、一覧取得の動きが理解しやすくなります。

アップロードする(upload_file と put_object)

ファイルを上げる:upload_file

open("report.csv", "w", encoding="utf-8").write("id,name\n1,ノート\n2,ペン\n")

r = s3.upload_file("report.csv", BUCKET, "reports/2026/report.csv")
print(r)

実行結果:

None

引数は「ローカルのパス、バケット名、キー」の順で、位置引数で渡せます。戻り値は None なので、成否は例外が出たかどうかで判断します。 公式リファレンスによると upload_file は大きなファイルを自動でマルチパート(分割)送信に切り替えます。

ContentTypeを指定する:ExtraArgs

upload_file は Content-Type を自動では付けません。何も指定しないと binary/octet-stream になり、ブラウザで開いたときに表示ではなくダウンロードになります。ExtraArgs で指定します。

s3.upload_file(
    "report.csv", BUCKET, "reports/2026/report_typed.csv",
    ExtraArgs={"ContentType": "text/csv; charset=utf-8"},
)
for k in ["reports/2026/report.csv", "reports/2026/report_typed.csv"]:
    h = s3.head_object(Bucket=BUCKET, Key=k)
    print(k, "|", h["ContentType"], "|", h["ContentLength"])

実行結果:

reports/2026/report.csv | binary/octet-stream | 29
reports/2026/report_typed.csv | text/csv; charset=utf-8 | 29

拡張子から種類を決めたいときは、標準ライブラリの mimetypes が使えます。

import mimetypes
print(mimetypes.guess_type("a.html"), mimetypes.guess_type("a.png"),
      mimetypes.guess_type("a.csv"), mimetypes.guess_type("a.unknownext"))
('text/html', None) ('image/png', None) ('text/csv', None) (None, None)

判定できない拡張子では None が返るので、guess_type(path)[0] or "application/octet-stream" のように既定値を添えておくと安心です。

メモリ上のデータを上げる:put_object

文字列や JSON をファイルに書き出さずに直接置けます。

import json

r = s3.put_object(
    Bucket=BUCKET, Key="notes/hello.txt",
    Body="こんにちは S3\n".encode("utf-8"),
    ContentType="text/plain; charset=utf-8",
)
print(r["ResponseMetadata"]["HTTPStatusCode"], r["ETag"])

s3.put_object(
    Bucket=BUCKET, Key="data/config.json",
    Body=json.dumps({"debug": False, "limit": 10}),
    ContentType="application/json",
)

実行結果:

200 "232e53b7b0c1ddb9e66e4a93b86ab1fe"

メタデータ(任意の付加情報)は Metadata に dict で渡し、head_object で読み戻せます。

s3.put_object(Bucket=BUCKET, Key="notes/meta.txt", Body=b"m",
              Metadata={"owner": "taro"}, ContentType="text/plain")
print(s3.head_object(Bucket=BUCKET, Key="notes/meta.txt")["Metadata"])
{'owner': 'taro'}

upload_file と put_object の選び方

観点 upload_file put_object
入力 ローカルファイルのパス bytes・str・ファイルオブジェクト
引数の渡し方 位置引数(Filename, Bucket, Key) キーワード引数(Bucket=, Key=, Body=)
大きなファイル 自動で分割送信 1回のリクエストで送る
ContentType など ExtraArgs={"ContentType": ...} ContentType=... を直接
戻り値 None ETag などを含む dict
向いている場面 ディスク上のファイル、サイズが大きいもの 生成した文字列や JSON、小さなデータ

開いたファイルや BytesIO をそのまま渡したいときは upload_fileobj もあります(s3.upload_fileobj(io.BytesIO(b"abc"), BUCKET, "notes/abc.bin") の形で、戻り値は upload_file と同じく None です)。

ダウンロードする(download_file と get_object)

ファイルに保存する:download_file

r = s3.download_file(BUCKET, "reports/2026/report.csv", "downloaded.csv")
print(r)
print(open("downloaded.csv", encoding="utf-8").read())

実行結果:

None
id,name
1,ノート
2,ペン

引数の順番は「バケット名、キー、保存先パス」で、upload_file とは並びが変わります。 ここを取り違えると、キーが見つからないという 404 になります(後述)。

中身を直接読む:get_object

r = s3.get_object(Bucket=BUCKET, Key="notes/hello.txt")
print(type(r["Body"]).__name__, r["ContentType"], r["ContentLength"])
print(repr(r["Body"].read().decode("utf-8")))

cfg = json.loads(s3.get_object(Bucket=BUCKET, Key="data/config.json")["Body"].read())
print(cfg)

実行結果:

StreamingChecksumBody text/plain; charset=utf-8 19
'こんにちは S3\n'
{'debug': False, 'limit': 10}

Body はストリームなので、read() は一度きりです。2回目は空になるため、使い回す値は変数に入れておきます。行ごとに処理したいときは iter_lines() が便利です。

for line in s3.get_object(Bucket=BUCKET, Key="reports/2026/report.csv")["Body"].iter_lines():
    print("line:", line.decode())
line: id,name
line: 1,ノート
line: 2,ペン

ファイルを作らずメモリに受けたい場合は download_fileobj に BytesIO を渡します。

import io
buf = io.BytesIO()
s3.download_fileobj(BUCKET, "notes/hello.txt", buf)
print(buf.getvalue())
b'\xe3\x81\x93\xe3\x82\x93\xe3\x81\xab\xe3\x81\xa1\xe3\x81\xaf S3\n'

一覧を取得する(list_objects_v2)

基本の形と、0件のときの受け方

r = s3.list_objects_v2(Bucket=BUCKET)
print("KeyCount", r["KeyCount"], "IsTruncated", r["IsTruncated"])
for o in r["Contents"]:
    print(f'{o["Key"]:32} {o["Size"]:>4} {o["LastModified"]:%Y-%m-%d %H:%M} {o["StorageClass"]}')

実行結果:

KeyCount 5 IsTruncated False
data/config.json                   29 2026-10-03 09:52 STANDARD
notes/hello.txt                    19 2026-10-03 09:52 STANDARD
notes/meta.txt                      1 2026-10-03 09:52 STANDARD
reports/2026/report.csv            29 2026-10-03 09:52 STANDARD
reports/2026/report_typed.csv      29 2026-10-03 09:52 STANDARD

キーの辞書順で返り、LastModified は UTC の datetime です。該当が0件のときは、応答に Contents キーそのものが入りません。

r = s3.list_objects_v2(Bucket=BUCKET, Prefix="none/")
print(sorted(k for k in r if k != "ResponseMetadata"), "KeyCount", r["KeyCount"])
try:
    for o in r["Contents"]:
        pass
except KeyError as e:
    print("KeyError:", e)
print("safe:", [o["Key"] for o in r.get("Contents", [])])

実行結果:

['EncodingType', 'IsTruncated', 'KeyCount', 'MaxKeys', 'Name', 'Prefix'] KeyCount 0
KeyError: 'Contents'
safe: []
直す前 直した後
for o in r["Contents"]: for o in r.get("Contents", []):

PrefixとDelimiterで「フォルダ」を取り出す

次のキーを追加して試します。

for k in ["photos/2025/a.jpg", "photos/2025/b.jpg", "photos/2026/c.jpg",
          "photos/readme.txt", "top.txt"]:
    s3.put_object(Bucket=BUCKET, Key=k, Body=b"x")

r = s3.list_objects_v2(Bucket=BUCKET, Delimiter="/")
print("直下のフォルダ:", [p["Prefix"] for p in r.get("CommonPrefixes", [])])
print("直下のファイル:", [o["Key"] for o in r.get("Contents", [])])

r = s3.list_objects_v2(Bucket=BUCKET, Prefix="photos/", Delimiter="/")
print("photos/ のフォルダ:", [p["Prefix"] for p in r.get("CommonPrefixes", [])])
print("photos/ のファイル:", [o["Key"] for o in r.get("Contents", [])])

r = s3.list_objects_v2(Bucket=BUCKET, Prefix="photos", Delimiter="/")
print("末尾スラッシュなし:", [p["Prefix"] for p in r.get("CommonPrefixes", [])],
      [o["Key"] for o in r.get("Contents", [])])

実行結果:

直下のフォルダ: ['data/', 'notes/', 'photos/', 'reports/']
直下のファイル: ['top.txt']
photos/ のフォルダ: ['photos/2025/', 'photos/2026/']
photos/ のファイル: ['photos/readme.txt']
末尾スラッシュなし: ['photos/'] []
指定 返るもの
Prefix だけ その文字列で始まるキーすべて(下の階層も含む)
Prefix と Delimiter="/" 直下のファイルは Contents、1段下のフォルダは CommonPrefixes
Delimiter だけ バケット直下のファイルとフォルダ
Prefix の末尾に "/" を付けない 前方一致なので、フォルダ名そのものが CommonPrefixes に1件入る

「あるフォルダの直下だけ」を見たいときは、Prefix の末尾にスラッシュを付けて Delimiter="/" を添える、と覚えておくと迷いません。

1000件を超える一覧をpaginatorで取る

list_objects_v2 は1回で最大1000件までしか返しません。2500件のキーを置いて確かめます。

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(16) as ex:
    list(ex.map(lambda i: s3.put_object(Bucket=BUCKET, Key=f"logs/2026/10/{i:05d}.log", Body=b"x"),
                range(2500)))

r = s3.list_objects_v2(Bucket=BUCKET, Prefix="logs/")
print("1回の呼び出し:", len(r["Contents"]), "IsTruncated", r["IsTruncated"],
      "NextContinuationToken あり:", "NextContinuationToken" in r)

実行結果:

1回の呼び出し: 1000 IsTruncated True NextContinuationToken あり: True

1000件で切れ、IsTruncated が True になりました。続きは paginator に任せるのが簡単です。

paginator = s3.get_paginator("list_objects_v2")
total = 0
for n, page in enumerate(paginator.paginate(Bucket=BUCKET, Prefix="logs/"), 1):
    c = page.get("Contents", [])
    total += len(c)
    print(f"page {n}: {len(c)} first={c[0]['Key']} last={c[-1]['Key']}")
print("total", total)

実行結果:

page 1: 1000 first=logs/2026/10/00000.log last=logs/2026/10/00999.log
page 2: 1000 first=logs/2026/10/01000.log last=logs/2026/10/01999.log
page 3: 500 first=logs/2026/10/02000.log last=logs/2026/10/02499.log
total 2500

キーだけを1つのリストにまとめる書き方と、条件で絞る search(JMESPath という問い合わせ記法)、件数を制御する PaginationConfig も動かしました。

keys = [o["Key"]
        for page in paginator.paginate(Bucket=BUCKET, Prefix="logs/")
        for o in page.get("Contents", [])]
print("keys", len(keys))

hit = list(paginator.paginate(Bucket=BUCKET, Prefix="logs/")
           .search("Contents[?ends_with(Key, `00.log`)].Key"))
print("search", len(hit), hit[:3])

pg = paginator.paginate(Bucket=BUCKET, Prefix="logs/",
                        PaginationConfig={"MaxItems": 1500, "PageSize": 500})
print("pages", [len(p["Contents"]) for p in pg])

実行結果:

keys 2500
search 25 ['logs/2026/10/00000.log', 'logs/2026/10/00100.log', 'logs/2026/10/00200.log']
pages [500, 500, 500]

paginator を使わず自分で回す場合は、NextContinuationToken を次の呼び出しの ContinuationToken に渡します。

token = None; cnt = 0; calls = 0
while True:
    kw = dict(Bucket=BUCKET, Prefix="logs/")
    if token:
        kw["ContinuationToken"] = token
    r = s3.list_objects_v2(**kw); calls += 1; cnt += r["KeyCount"]
    if not r["IsTruncated"]:
        break
    token = r["NextContinuationToken"]
print("calls", calls, "count", cnt)
calls 3 count 2500
方法 コード量 向いている場面
paginator 少ない ふだんはこれで十分
search つき paginator 少ない 拡張子やサイズで絞りたい
ContinuationToken を自分で回す 多い 途中のトークンを保存して後で再開したい

なお search の絞り込みは受け取った後の処理です。S3 側で絞れるのは Prefix だけなので、よく絞る条件はキーの先頭側に置く設計が一覧の速さにつながります。

存在を確認する(head_object)

head_object は本文を取らずに情報だけを返すので、存在確認に向いています。無いときは例外になるため、関数にまとめます。

from botocore.exceptions import ClientError

def exists(bucket, key):
    try:
        s3.head_object(Bucket=bucket, Key=key)
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] in ("404", "NoSuchKey", "NotFound"):
            return False
        raise          # 権限や通信の問題は握りつぶさず上に伝える

print(exists(BUCKET, "notes/hello.txt"), exists(BUCKET, "no/such.txt"))

実行結果:

True False

head_object が返す主なキーは ContentLength(サイズ)、ContentType、ETag、LastModified、Metadata です。「無ければ作る」という流れなら、存在確認を挟まず get_object を呼んで NoSuchKey を捕まえるほうが、通信が1回で済みます。

コピー・移動・削除する

コピーと移動

s3.copy({"Bucket": BUCKET, "Key": "notes/hello.txt"}, BUCKET, "backup/hello.txt")
r = s3.copy_object(Bucket=BUCKET, Key="backup/hello2.txt",
                   CopySource={"Bucket": BUCKET, "Key": "notes/hello.txt"})
print(r["CopyObjectResult"]["ETag"])
print([o["Key"] for o in s3.list_objects_v2(Bucket=BUCKET, Prefix="backup/")["Contents"]])

# 移動(名前の変更)は、コピーしてから元を消す
s3.copy({"Bucket": BUCKET, "Key": "top.txt"}, BUCKET, "archive/top.txt")
s3.delete_object(Bucket=BUCKET, Key="top.txt")

実行結果:

"232e53b7b0c1ddb9e66e4a93b86ab1fe"
['backup/hello.txt', 'backup/hello2.txt']

S3 に「移動」や「名前の変更」という操作はないため、コピーと削除の2手で行います。copy は upload_file と同じ仕組みで動く高水準のメソッド、copy_object は API を1回呼ぶ低水準のメソッドです。

1件ずつ消す・まとめて消す

r = s3.delete_object(Bucket=BUCKET, Key="backup/hello2.txt")
print(r["ResponseMetadata"]["HTTPStatusCode"])
r = s3.delete_object(Bucket=BUCKET, Key="backup/not-exist.txt")
print("無いキー:", r["ResponseMetadata"]["HTTPStatusCode"])

r = s3.delete_objects(Bucket=BUCKET, Delete={
    "Objects": [{"Key": "photos/2025/a.jpg"}, {"Key": "photos/2025/b.jpg"}],
})
print({k: v for k, v in r.items() if k != "ResponseMetadata"})

実行結果:

204
無いキー: 204
{'Deleted': [{'Key': 'photos/2025/a.jpg'}, {'Key': 'photos/2025/b.jpg'}]}

存在しないキーを消しても例外にはなりません。 公式の API リファレンスにも、指定したオブジェクトが見つからない場合は削除済みとして結果を返すと書かれています。

delete_objects は1回のリクエストで1000キーまで、というのが公式リファレンスの決まりです。先ほどの2500件は1000件ずつに分けて消します。

for i in range(0, len(keys), 1000):
    chunk = keys[i:i + 1000]
    r = s3.delete_objects(Bucket=BUCKET,
                          Delete={"Objects": [{"Key": k} for k in chunk], "Quiet": True})
    print("batch", i // 1000 + 1, "errors", r.get("Errors", []))
print("残り", s3.list_objects_v2(Bucket=BUCKET, Prefix="logs/")["KeyCount"])

実行結果:

batch 1 errors []
batch 2 errors []
batch 3 errors []
残り 0

"Quiet": True を付けると、応答にはエラーになったキーだけが入ります。大量に消すときは Errors が空かどうかを見れば足ります。

署名付きURLを発行する(generate_presigned_url)

署名付き URL は、AWS の認証情報を持たない相手に、期限つきで1つのオブジェクトの取得やアップロードを許すための URL です。

import urllib.request
from botocore.config import Config

s4 = boto3.client("s3", config=Config(signature_version="s3v4"))

url = s4.generate_presigned_url(
    "get_object",
    Params={"Bucket": BUCKET, "Key": "notes/hello.txt"},
    ExpiresIn=600,                      # 秒。ここでは10分
)
print(url)
print(urllib.request.urlopen(url).read().decode())

実行結果(ホスト名の部分は実際の AWS の形に置き換えています):

https://rjc-sample-bucket.s3.ap-northeast-1.amazonaws.com/notes/hello.txt?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=test%2F20261003%2Fap-northeast-1%2Fs3%2Faws4_request&X-Amz-Date=20261003T095227Z&X-Amz-Expires=600&X-Amz-SignedHeaders=host&X-Amz-Signature=e3422c8d071c7bf1249de30a6e7674f780150e40e4612598bdee4c7ba055cca8
こんにちは S3

アップロード用の URL も同じ要領で作れます。ContentType を Params に入れた場合は、送る側も同じヘッダーを付けます。

url2 = s4.generate_presigned_url(
    "put_object",
    Params={"Bucket": BUCKET, "Key": "uploads/from-url.txt", "ContentType": "text/plain"},
    ExpiresIn=600,
)
req = urllib.request.Request(url2, data=b"via presigned url", method="PUT",
                             headers={"Content-Type": "text/plain"})
print("PUT status", urllib.request.urlopen(req).status)
print(s3.get_object(Bucket=BUCKET, Key="uploads/from-url.txt")["Body"].read())

実行結果:

PUT status 200
b'via presigned url'

知っておきたい性質を表にまとめます。

項目 内容
発行時の通信 行わない。手元で署名を計算して文字列を返すだけ
存在しないキー URL は発行できる。開いたときに 404(NoSuchKey)が返る
誰の権限で動くか URL を発行した IAM の権限。発行者に s3:GetObject が無ければ開けない
有効期限の上限 AWS CLI や SDK で発行する場合は最長7日(公式ユーザーガイド)
一時的な認証情報で発行 指定した期限と認証情報の期限の、早いほうで切れる(同上)

存在しないキーの URL を開いたときの応答も確かめました。

import urllib.error
u = s4.generate_presigned_url("get_object",
                              Params={"Bucket": BUCKET, "Key": "no/such.txt"}, ExpiresIn=600)
try:
    urllib.request.urlopen(u)
except urllib.error.HTTPError as e:
    print(e.code, e.read()[:110])
404 b'<?xml version="1.0" encoding="utf-8"?>\n<Error><Code>NoSuchKey</Code><Message>The specified key does not exist.'

エラーをコード別に捕まえる(ClientError)

boto3 のエラーの多くは botocore.exceptions.ClientError として届き、e.response["Error"]["Code"] に種類が入ります。操作ごとに何が届くかを実際に出して一覧にしました。

操作ごとの例外とエラーコード一覧

操作と状況 例外クラス Code HTTP
get_object:キーが無い NoSuchKey(ClientError の子) NoSuchKey 404
head_object:キーが無い ClientError 404 404
download_file:キーが無い ClientError 404 404
get_object / put_object / list_objects_v2:バケットが無い NoSuchBucket(ClientError の子) NoSuchBucket 404
head_bucket:バケットが無い ClientError 404 404
upload_file:バケットが無い boto3.exceptions.S3UploadFailedError (メッセージ内に NoSuchBucket) -
upload_file:ローカルファイルが無い FileNotFoundError - -
put_object / list_objects_v2 / delete_object:権限が無い ClientError AccessDenied 403
create_bucket:自分の同名バケットがある BucketAlreadyOwnedByYou BucketAlreadyOwnedByYou 409
delete_bucket:中身が残っている ClientError BucketNotEmpty 409
必須の引数が足りない botocore.exceptions.ParamValidationError - -

実際のメッセージは次のとおりです。

get_object(キーなし)
    An error occurred (NoSuchKey) when calling the GetObject operation: The specified key does not exist.
head_object(キーなし)
    An error occurred (404) when calling the HeadObject operation: Not Found
download_file(キーなし)
    An error occurred (404) when calling the HeadObject operation: Not Found
get_object(バケットなし)
    An error occurred (NoSuchBucket) when calling the GetObject operation: The specified bucket does not exist
upload_file(バケットなし)
    Failed to upload report.csv to rjc-no-such-bucket/a.csv: An error occurred (NoSuchBucket) when calling the PutObject operation: The specified bucket does not exist
get_object(Key を渡し忘れ)
    Parameter validation failed:
    Missing required parameter in input: "Key"

ここから読み取れるポイントは3つです。

  • head_object と download_file は、キーが無いとき NoSuchKey ではなく "404" というコードになります。 HEAD リクエストの応答には本文が無く、エラー名を運べないためです。download_file は内部で先に HeadObject を呼ぶので、同じ 404 になります。
  • upload_file だけは ClientError ではなく S3UploadFailedError が届きます。 except ClientError だけでは受けられないので、両方を並べます。
  • 引数の不足や型の違いは、通信の前に ParamValidationError になります。

コード別に分岐する書き方

def read_text(bucket, key):
    try:
        return s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode("utf-8")
    except ClientError as e:
        code = e.response["Error"]["Code"]
        if code == "NoSuchKey":
            print(f"[skip] キーがありません: {key}")
            return None
        if code == "NoSuchBucket":
            print(f"[stop] バケットがありません: {bucket}")
            raise
        if code in ("AccessDenied", "403"):
            print(f"[stop] 権限がありません: s3:GetObject / {bucket}/{key}")
            raise
        raise

print(repr(read_text(BUCKET, "notes/hello.txt")))
print(read_text(BUCKET, "no/such.txt"))
try:
    read_text("rjc-no-such-bucket", "a")
except ClientError as e:
    print("re-raised:", e.response["Error"]["Code"])

実行結果:

'こんにちは S3\n'
[skip] キーがありません: no/such.txt
None
[stop] バケットがありません: rjc-no-such-bucket
re-raised: NoSuchBucket

「無いのは想定内なので None を返す」「設定の誤りなので止める」を、コードごとに分けるのが要点です。最後の raise を忘れずに置くと、想定していないエラーが黙って消えることを防げます。

s3.exceptions で捕まえる書き方と、その範囲

client には s3.exceptions.NoSuchKey のような例外クラスも用意されています。

try:
    s3.get_object(Bucket=BUCKET, Key="no/such.txt")
except s3.exceptions.NoSuchKey as e:
    print("get_object は NoSuchKey で受けられる:", e.response["Error"]["Code"])

try:
    s3.head_object(Bucket=BUCKET, Key="no/such.txt")
except s3.exceptions.NoSuchKey:
    print("head_object も NoSuchKey で受けられる")
except ClientError as e:
    print("head_object は NoSuchKey では受けられない。Code =", e.response["Error"]["Code"])

実行結果:

get_object は NoSuchKey で受けられる: NoSuchKey
head_object は NoSuchKey では受けられない。Code = 404

読みやすい書き方ですが、head_object の 404 のように当てはまらない操作があります。迷ったら ClientError で受けて Code を見る書き方に統一すると、どの操作でも同じ形で扱えます。

AccessDenied(権限が無い)のとき

s3:GetObject だけを許可した IAM ユーザーで各操作を試した結果です。

get_object       -> OK
put_object       -> Code='AccessDenied' HTTP=403
    An error occurred (AccessDenied) when calling the PutObject operation: Access Denied
list_objects_v2  -> Code='AccessDenied' HTTP=403
    An error occurred (AccessDenied) when calling the ListObjectsV2 operation: Access Denied
delete_object    -> Code='AccessDenied' HTTP=403
    An error occurred (AccessDenied) when calling the DeleteObject operation: Access Denied
upload_file      -> S3UploadFailedError
    Failed to upload report.csv to rjc-auth-bucket/c.csv: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied

メッセージの「when calling the ○○ operation」の部分が、足りない権限の手がかりです。

呼んだメソッド 必要な権限(IAM のアクション) 対象
get_object / download_file / head_object s3:GetObject オブジェクト(arn:aws:s3:::バケット名/*)
put_object / upload_file s3:PutObject オブジェクト
delete_object / delete_objects s3:DeleteObject オブジェクト
list_objects_v2 s3:ListBucket バケット(arn:aws:s3:::バケット名)

一覧の権限だけはバケット自体に付ける、という点がつまずきやすいところです。また公式の API リファレンスには、s3:ListBucket の権限が無い状態で存在しないキーに head_object すると、404 ではなく 403 が返ると書かれています。存在確認の関数で 403 を「無い」と扱わず、上に伝える書き方にしておくのはこのためです。

clientとresourceの違い

boto3 には boto3.client("s3") と boto3.resource("s3") の2つの入口があります。resource はオブジェクト指向の書き方ができます。

res = boto3.resource("s3")
b = res.Bucket(BUCKET)
print([o.key for o in b.objects.filter(Prefix="notes/")])

obj = res.Object(BUCKET, "notes/hello.txt")
print(obj.content_type, obj.content_length, obj.get()["Body"].read().decode().strip())

# resource の中から client を取り出すこともできる
print(res.meta.client.head_object(Bucket=BUCKET, Key="notes/hello.txt")["ContentLength"])

実行結果:

['notes/hello.txt', 'notes/meta.txt']
text/plain; charset=utf-8 19 こんにちは S3
19
観点 client resource
作り方 boto3.client("s3") boto3.resource("s3")
書き方 API と1対1のメソッド。引数も戻り値も dict Bucket や Object といったオブジェクトを操作
一覧のページ送り paginator か ContinuationToken objects.filter() や objects.all() が自動で続きを取る
新しい機能の追加 追加され続ける 追加されない(公式ガイドに明記。既存の機能は引き続き動く)
スレッドでの扱い この記事では client を1つ作り、16スレッドから put_object を呼んで2500件を置いた 公式ガイドは、スレッド間で共有せずスレッドごとに作るよう案内
例外 ClientError ClientError(同じ)
向いている場面 これから書くコード全般 既にあるコードの保守、短いスクリプト

公式ガイドには「boto3 の resource インターフェイスに新機能を追加する予定はなく、新しいサービス機能は client から使える」という趣旨の注記があります。これから書くコードは client を基本にすると、後から機能を足したくなったときにも同じ書き方のまま進められます。resource で書かれたコードに client の機能を足したいときは、上の例のように res.meta.client を使うと書き換えずに済みます。

つまずきやすい点と直し方

症状 原因 直し方
一覧が1000件で止まる list_objects_v2 の1回の上限 paginator を使う
KeyError: 'Contents' 0件のとき Contents キーが無い r.get("Contents", [])
download_file で 404 引数の順(バケット、キー、保存先)の取り違えやキーの打ち間違い 順番とキーの文字列を見直す
ブラウザで開くとダウンロードになる ContentType が binary/octet-stream ExtraArgs={"ContentType": ...} を付ける
upload_file の例外が except ClientError を通り抜ける S3UploadFailedError は別の系統 except (ClientError, S3UploadFailedError)
head_object の NoSuchKey が捕まらない コードが "404" Code に "404" を含めて判定
Body.read() の2回目が空 ストリームは一度きり 変数に入れて使い回す
フォルダ直下だけ欲しいのに全部返る Delimiter の指定なし Prefix="dir/", Delimiter="/"

よくある質問

Q. upload_file と put_object はどちらを使えばよいですか? A. ディスク上のファイルを上げるなら upload_file、プログラムの中で作った文字列や JSON を置くなら put_object が向いています。upload_file は大きなファイルを自動で分割して送るので、サイズが読めないときも安心です。

Q. resource で書かれたコードは client に書き直す必要がありますか? A. 既存の機能は引き続き動くと公式ガイドに書かれているので、急いで書き直す必要はありません。新しい機能を使いたい箇所だけ res.meta.client で client を取り出して呼ぶ、という進め方ができます。

Q. フォルダを作るにはどうしますか? A. フォルダを作る操作は必要ありません。photos/2026/c.jpg のようにスラッシュ入りのキーでオブジェクトを置けば、一覧で Delimiter="/" を指定したときにフォルダとして見えます。

Q. 署名付き URL の期限はどのくらいにすればよいですか? A. 用途に足りる最短の時間にするのが基本です。その場でダウンロードしてもらうなら数分から十数分、という決め方が扱いやすいです。SDK で発行する場合の上限は7日です。

まとめ

  • 入口は boto3.client("s3")。 アップロードは upload_file か put_object、ダウンロードは download_file か get_object を使い分ける
  • upload_file は ContentType を自動で付けない。 ExtraArgs={"ContentType": ...} で指定する
  • list_objects_v2 は1回1000件まで。 それ以上は paginator、0件のときは r.get("Contents", []) で受ける
  • フォルダ直下の一覧は Prefix="dir/", Delimiter="/"。 下位フォルダは CommonPrefixes に入る
  • エラーは ClientError の Code で分岐する。 get_object は NoSuchKey、head_object と download_file は "404"、upload_file は S3UploadFailedError
  • AccessDenied はメッセージの操作名から必要な権限を特定する。 一覧の s3:ListBucket はバケット自体に付ける
  • 新しく書くコードは client を基本にする。 resource には新機能が追加されない

参考・出典

  • Boto3 ドキュメント「Resources」(resource への新機能追加についての注記、スレッドでの扱い)
  • Boto3 ドキュメント「S3.Client.upload_file」(引数、ExtraArgs、マルチパート)
  • Amazon S3 API リファレンス「DeleteObjects」(1回1000キーまで、Quiet、存在しないキーの扱い)
  • Amazon S3 API リファレンス「HeadObject」(必要な権限、s3:ListBucket の有無による 404 と 403)
  • Amazon S3 ユーザーガイド「署名付き URL を使用したオブジェクトのダウンロードおよびアップロード」(有効期限の上限、発行者の権限)

本記事の内容は、ご自由にお使いください。
ご利用の際は、出典として本ページへのリンクを記載いただけますようお願いします。

(記載例)出典:株式会社RJC「boto3でS3を操作する方法|アップロード・ダウンロード・一覧取得」

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。

WE ARE HIRING

RJCで一緒に開発しながら、
成長を楽しみませんか?

RJCは、Web・モバイル・AIを活用した開発プロジェクトで、テックリードやPM・PMOも活躍するシステム開発会社です。会社を知る、待遇を確かめる、話を聞いてみる。気になるところ見てみてください!

  • 127日年間休日
  • 12時間平均残業時間
  • 毎日ガチャ遊びココロも大切にする福利厚生。アマギフなどの賞品ラインナップ!

ほかにも、チケットレストラン、書籍読み放題、2年ごとの慰労報奨(休暇 or 金一封)、11期連続の黒字決算。

ABOUT RJC RJCがどんな会社か知る 考え方、研修、働き方、福利厚生、社員の前職まで。RJCのことが丸わかり! RJC丸わかりページへ JOB DESCRIPTION 仕事内容・待遇を見てみる 仕事内容、給与・待遇、選考の流れ。経験者も未経験も!応募前に知りたいこと、まとめました! 募集要項を見る ENTRY エントリーする エントリーは1〜2分・履歴書不要です。まずは話を聞いてみたい、という方でも歓迎です! エントリーフォームへ

RJCで一緒に開発しながら、 成長を楽しみませんか?