boto3のエラーハンドリング｜ClientErrorのコード取得とリトライ設定

boto3 のエラーは、AWS が返したエラー（botocore.exceptions.ClientError）と、リクエストを送る前後に手元で起きるエラー（BotoCoreError の仲間）の 2 系統に分かれます。ClientError は e.response["Error"]["Code"] でエラーコードを取り出して分岐するのが基本形です。

この記事では、e.response の中身を実際の辞書で確かめ、client.exceptions.XXX との使い分け、サービス別のエラーコード、botocore.exceptions の発生条件を表にします。後半では Config(retries=...) の 3 つのモードで「何回送られるか」を数え、リトライされるエラーとされないエラー、自前のバックオフと重ねたときの回数までコードで確かめます。

確認環境：Ubuntu 24.04、Python 3.11.15、boto3 1.43.103（botocore 1.43.103）。AWS の API をローカルで再現するツール（moto 5.2.3）で実行しました。S3 などを呼ぶコードは、実際の AWS でそのまま使える形（接続先の指定なし）で載せています。リトライの回数を数える節だけは、決まった応答を返す検証用サーバーに接続します。

boto3エラーハンドリングの早見表

ClientError と BotoCoreError は親子関係にありません。 両方を受けたいときは、except を 2 つ書きます。

ClientErrorのe.responseの中身を見る

以降のコードは、rjc-demo-bucket というバケット（中にオブジェクトが 1 つある）を用意した状態で動かしています。存在しないキーを get_object して、例外の中身を表示します。

# err_response.py
import pprint
import boto3
from botocore.exceptions import ClientError

s3 = boto3.client("s3")
try:
    s3.get_object(Bucket="rjc-demo-bucket", Key="report.csv")
except ClientError as e:
    print("例外クラス :", type(e).__name__)
    print("str(e)     :", e)
    print("操作名     :", e.operation_name)
    print("Code       :", e.response["Error"]["Code"])
    print("Message    :", e.response["Error"]["Message"])
    meta = e.response["ResponseMetadata"]
    print("HTTPStatus :", meta["HTTPStatusCode"])
    print("RequestId  :", meta["RequestId"])
    print("RetryAttempts:", meta["RetryAttempts"])
    print("--- e.response 全体（HTTPHeaders は長いので省略して表示）")
    shown = dict(e.response)
    shown["ResponseMetadata"] = {**meta, "HTTPHeaders": "…（省略）…"}
    pprint.pprint(shown)

例外クラス : NoSuchKey
str(e)     : An error occurred (NoSuchKey) when calling the GetObject operation: The specified key does not exist.
操作名     : GetObject
Code       : NoSuchKey
Message    : The specified key does not exist.
HTTPStatus : 404
RequestId  : pEPoQ6Y09IvPuoKHiFJEu6OhHgbEtbaHbRqy9OM9zm4yj0AinJSf
RetryAttempts: 0
--- e.response 全体（HTTPHeaders は長いので省略して表示）
{'Error': {'Code': 'NoSuchKey',
           'Key': 'report.csv',
           'Message': 'The specified key does not exist.'},
 'ResponseMetadata': {'HTTPHeaders': '…（省略）…',
                      'HTTPStatusCode': 404,
                      'RequestId': 'pEPoQ6Y09IvPuoKHiFJEu6OhHgbEtbaHbRqy9OM9zm4yj0AinJSf',
                      'RetryAttempts': 0}}

公式ドキュメントの例では、ResponseMetadata に HostId というキーも載っています。どのサービスでも必ずあるのは Error.Code と ResponseMetadataと考え、追加のキーは .get() で読むと安心です。

client.exceptionsとClientError＋コード判定の使い分け

捕まえ方は 2 通りあります。

# two_ways.py
import boto3
from botocore.exceptions import ClientError

s3 = boto3.client("s3")

# 方法1: client.exceptions.XXX で、決まった 1 種類だけを捕まえる
try:
    s3.get_object(Bucket="rjc-demo-bucket", Key="report.csv")
except s3.exceptions.NoSuchKey as e:
    print("方法1 NoSuchKey を捕まえた:", e.response["Error"]["Message"])

# 方法2: ClientError で受けて、Error.Code で分岐する
def read_or_none(bucket, key):
    try:
        return s3.get_object(Bucket=bucket, Key=key)["Body"].read()
    except ClientError as e:
        code = e.response["Error"]["Code"]
        if code in ("NoSuchKey", "404"):
            return None                      # 無いのは想定内
        if code == "NoSuchBucket":
            raise RuntimeError(f"バケット {bucket} がありません") from e
        raise                                # それ以外は上に伝える

print("方法2 無いキー      :", read_or_none("rjc-demo-bucket", "report.csv"))
try:
    read_or_none("rjc-none-bucket", "report.csv")
except RuntimeError as e:
    print("方法2 無いバケット  :", e)

# client.exceptions に無いコードは、ClientError でしか受けられない
for label, call in [
    ("head_object（無いキー）", lambda: s3.head_object(Bucket="rjc-demo-bucket", Key="report.csv")),
    ("delete_bucket（空でない）", lambda: s3.delete_bucket(Bucket="rjc-demo-bucket")),
]:
    try:
        call()
    except s3.exceptions.NoSuchKey:
        print(label, "-> NoSuchKey クラス")
    except ClientError as e:
        print(label, "->", type(e).__name__, "/ Code =", e.response["Error"]["Code"])

# client.exceptions に用意されているクラスの数
for name in ["s3", "dynamodb", "sqs", "iam", "sts"]:
    c = boto3.client(name)
    names = [n for n in dir(c.exceptions) if n[0].isupper() and n != "ClientError"]
    print(f"{name:9} {len(names):3} クラス 例: {', '.join(names[:3])}")

方法1 NoSuchKey を捕まえた: The specified key does not exist.
方法2 無いキー      : None
方法2 無いバケット  : バケット rjc-none-bucket がありません
head_object（無いキー） -> ClientError / Code = 404
delete_bucket（空でない） -> ClientError / Code = BucketNotEmpty
s3         20 クラス 例: AccessDenied, AnnotationLimitExceeded, AnnotationNameTooLong
dynamodb   34 クラス 例: BackupInUseException, BackupNotFoundException, ConditionalCheckFailedException
sqs        28 クラス 例: BatchEntryIdsNotDistinct, BatchRequestTooLong, EmptyBatchRequest
iam        38 クラス 例: AccountNotManagementOrDelegatedAdministratorException, CallerIsNotManagementAccountException, ConcurrentModificationException
sts        12 クラス 例: ExpiredTokenException, ExpiredTradeInTokenException, IDPCommunicationErrorException

出力の 4〜5 行目がポイントです。head_object で無いキーを指定したときのコードは "404"、空でないバケットを消そうとしたときは BucketNotEmpty で、どちらも client.exceptions にクラスが無く、素の ClientError として届きます。 client.exceptions に用意されているのは、API 定義に載っているエラーだけです。

公式ドキュメントにも、ClientError で受けてエラーコードを調べる方法が、サービス側のすべてのエラーを捕まえる最も確実な方法だと書かれています。迷ったら右側の書き方を基本にして、読みやすさが欲しい箇所だけ左側を使うとよいでしょう。

なお、client.exceptions のクラスは ClientError の子クラスです。except ClientError を先に書くと、後ろの except s3.exceptions.NoSuchKey には届かないので、細かいほうを先に書きます。

サービス別：よく出るエラーコード一覧

次のスクリプトで、5 つのサービスのエラーを実際に起こし、例外クラス名・Error.Code・HTTP ステータスを並べました。

# codes.py
import boto3
from botocore.exceptions import ClientError

s3, ddb, sqs, iam = (boto3.client(n) for n in ["s3", "dynamodb", "sqs", "iam"])
key = [{"AttributeName": "id", "KeyType": "HASH"}]
attr = [{"AttributeName": "id", "AttributeType": "S"}]
ddb.create_table(TableName="Tasks", KeySchema=key, AttributeDefinitions=attr, BillingMode="PAY_PER_REQUEST")
ddb.put_item(TableName="Tasks", Item={"id": {"S": "1"}})
queue_url = sqs.create_queue(QueueName="tasks")["QueueUrl"]
iam.create_user(UserName="taro")

cases = [
    ("s3.get_object 無いキー", lambda: s3.get_object(Bucket="rjc-demo-bucket", Key="x")),
    ("s3.head_object 無いキー", lambda: s3.head_object(Bucket="rjc-demo-bucket", Key="x")),
    ("s3.get_object 無いバケット", lambda: s3.get_object(Bucket="rjc-none-bucket", Key="x")),
    ("s3.create_bucket 作成済み", lambda: s3.create_bucket(Bucket="rjc-demo-bucket", CreateBucketConfiguration={"LocationConstraint": "ap-northeast-1"})),
    ("s3.delete_bucket 空でない", lambda: s3.delete_bucket(Bucket="rjc-demo-bucket")),
    ("ddb.get_item 無いテーブル", lambda: ddb.get_item(TableName="Nothing", Key={"id": {"S": "1"}})),
    ("ddb.create_table 作成済み", lambda: ddb.create_table(TableName="Tasks", KeySchema=key, AttributeDefinitions=attr, BillingMode="PAY_PER_REQUEST")),
    ("ddb.put_item 条件に合わない", lambda: ddb.put_item(TableName="Tasks", Item={"id": {"S": "1"}}, ConditionExpression="attribute_not_exists(id)")),
    ("ddb.get_item キー名が違う", lambda: ddb.get_item(TableName="Tasks", Key={"idx": {"S": "1"}})),
    ("sqs.get_queue_url 無いキュー", lambda: sqs.get_queue_url(QueueName="nothing")),
    ("sqs.create_queue 属性違いで再作成", lambda: sqs.create_queue(QueueName="tasks", Attributes={"VisibilityTimeout": "99"})),
    ("sqs.delete_message 受信ハンドルが不正", lambda: sqs.delete_message(QueueUrl=queue_url, ReceiptHandle="bogus")),
    ("iam.get_role 無いロール", lambda: iam.get_role(RoleName="nothing")),
    ("iam.create_user 作成済み", lambda: iam.create_user(UserName="taro")),
    ("iam.create_policy JSON が不正", lambda: iam.create_policy(PolicyName="p", PolicyDocument="{bad")),
]
for label, call in cases:
    try:
        call()
    except ClientError as e:
        err, status = e.response["Error"], e.response["ResponseMetadata"]["HTTPStatusCode"]
        print(f"{label}\n    class={type(e).__name__}  Code={err['Code']}  HTTP={status}")

s3.get_object 無いキー
    class=NoSuchKey  Code=NoSuchKey  HTTP=404
s3.head_object 無いキー
    class=ClientError  Code=404  HTTP=404
s3.get_object 無いバケット
    class=NoSuchBucket  Code=NoSuchBucket  HTTP=404
s3.create_bucket 作成済み
    class=BucketAlreadyOwnedByYou  Code=BucketAlreadyOwnedByYou  HTTP=409
s3.delete_bucket 空でない
    class=ClientError  Code=BucketNotEmpty  HTTP=409
ddb.get_item 無いテーブル
    class=ResourceNotFoundException  Code=ResourceNotFoundException  HTTP=400
ddb.create_table 作成済み
    class=ResourceInUseException  Code=ResourceInUseException  HTTP=400
ddb.put_item 条件に合わない
    class=ConditionalCheckFailedException  Code=ConditionalCheckFailedException  HTTP=400
ddb.get_item キー名が違う
    class=ClientError  Code=ValidationException  HTTP=400
sqs.get_queue_url 無いキュー
    class=QueueDoesNotExist  Code=AWS.SimpleQueueService.NonExistentQueue  HTTP=400
sqs.create_queue 属性違いで再作成
    class=QueueNameExists  Code=QueueAlreadyExists  HTTP=400
sqs.delete_message 受信ハンドルが不正
    class=ReceiptHandleIsInvalid  Code=ReceiptHandleIsInvalid  HTTP=400
iam.get_role 無いロール
    class=NoSuchEntityException  Code=NoSuchEntity  HTTP=404
iam.create_user 作成済み
    class=EntityAlreadyExistsException  Code=EntityAlreadyExists  HTTP=409
iam.create_policy JSON が不正
    class=MalformedPolicyDocumentException  Code=MalformedPolicyDocument  HTTP=400

クラス名と Error.Code が一致しないサービスがあります。 IAM はクラス名が NoSuchEntityException、コードが NoSuchEntity です。SQS はクラス名が QueueDoesNotExist、コードが AWS.SimpleQueueService.NonExistentQueue です。コードで分岐するときは、クラス名から推測せず、一度出力して確かめるのが確実です。

実行して確かめたものに、公式ドキュメントと botocore 同梱の API 定義から読み取ったものを足した一覧です（「確認」の列が「実行」のものは上の出力のとおり）。

S3

DynamoDB

「再試行」の列は、公式ドキュメントのエラー表にある「OK to retry?」の値です。

SQS

API 定義には、このほか次のクラスがあります（説明は API 定義の文面から）。

IAM

STS

各サービスの全コードは、公式ドキュメントが案内しているとおり、サービスごとの API リファレンスにあるエラーの節で確認できます。

botocore.exceptionsの発生条件一覧

ここからは、AWS に届く前後に手元で起きるエラーです。認証情報もリージョンも設定していない状態を作り、順に起こしました。

# botocore_errors.py
import boto3
import botocore.exceptions as be
from botocore.config import Config

def show(label, func):
    try:
        func()
    except be.ClientError as e:
        print(f"[{label}] ClientError: {e}")
    except be.BotoCoreError as e:
        parents = [c.__name__ for c in type(e).__mro__[1:] if c.__module__ == "botocore.exceptions"]
        print(f"[{label}] {type(e).__name__}（親: {' > '.join(parents)}）\n    {str(e).strip()[:110]}")

fast = Config(connect_timeout=1, read_timeout=1, retries={"total_max_attempts": 1})

show("引数の型が違う", lambda: boto3.client("s3", region_name="ap-northeast-1").list_objects_v2(Bucket="rjc-demo-bucket", MaxKeys="ten"))
show("必須の引数が無い", lambda: boto3.client("s3", region_name="ap-northeast-1").get_object(Bucket="rjc-demo-bucket"))
show("認証情報が無い", lambda: boto3.client("s3", region_name="ap-northeast-1").list_buckets())
show("認証情報が片方だけ", lambda: boto3.client("s3", region_name="ap-northeast-1", aws_access_key_id="AKIAEXAMPLE").list_buckets())
show("リージョンが無い", lambda: boto3.client("dynamodb"))
show("profile が無い", lambda: boto3.Session(profile_name="nothing"))
show("サービス名が違う", lambda: boto3.client("s33", region_name="ap-northeast-1"))
show("接続先に届かない", lambda: boto3.client(
    "s3", region_name="ap-northeast-1", endpoint_url="http://localhost:9219",
    aws_access_key_id="x", aws_secret_access_key="x", config=fast).list_buckets())
show("ページ分割に非対応の操作", lambda: boto3.client("s3", region_name="ap-northeast-1").get_paginator("get_object"))
show("retries の指定が違う", lambda: boto3.client("s3", region_name="ap-northeast-1", config=Config(retries={"mode": "fast"})))

設定を空にするため、次のように実行しています。

env -u AWS_ACCESS_KEY_ID -u AWS_SECRET_ACCESS_KEY -u AWS_DEFAULT_REGION \
    AWS_CONFIG_FILE=./empty AWS_SHARED_CREDENTIALS_FILE=./empty \
    AWS_EC2_METADATA_DISABLED=true python3 botocore_errors.py

[引数の型が違う] ParamValidationError（親: BotoCoreError）
    Parameter validation failed:
Invalid type for parameter MaxKeys, value: ten, type: <class 'str'>, valid types:
[必須の引数が無い] ParamValidationError（親: BotoCoreError）
    Parameter validation failed:
Missing required parameter in input: "Key"
[認証情報が無い] NoCredentialsError（親: BotoCoreError）
    Unable to locate credentials
[認証情報が片方だけ] PartialCredentialsError（親: BotoCoreError）
    Partial credentials found in explicit, missing: aws_secret_access_key
[リージョンが無い] NoRegionError（親: BaseEndpointResolverError > BotoCoreError）
    You must specify a region.
[profile が無い] ProfileNotFound（親: BotoCoreError）
    The config profile (nothing) could not be found
[サービス名が違う] UnknownServiceError（親: DataNotFoundError > BotoCoreError）
    Unknown service: 's33'. Valid service names are: accessanalyzer, account, account-access, acm, acm-pca, agent-
[接続先に届かない] EndpointConnectionError（親: ConnectionError > BotoCoreError）
    Could not connect to the endpoint URL: "http://localhost:9219/"
[ページ分割に非対応の操作] OperationNotPageableError（親: BotoCoreError）
    Operation cannot be paginated: get_object
[retries の指定が違う] InvalidRetryModeError（親: InvalidRetryConfigurationError > BotoCoreError）
    Invalid value provided to "mode": "fast" must be one of: ('legacy', 'standard', 'adaptive')

待機（waiter）が時間内に終わらないときは WaiterError です。

# waiter_error.py
import boto3
from botocore.exceptions import WaiterError

s3 = boto3.client("s3")
try:
    s3.get_waiter("bucket_exists").wait(
        Bucket="rjc-none-bucket", WaiterConfig={"Delay": 1, "MaxAttempts": 3})
except WaiterError as e:
    print(type(e).__name__, ":", e)
    print("最後の応答の Code:", e.last_response["Error"]["Code"])

WaiterError : Waiter BucketExists failed: Max attempts exceeded. Previously accepted state: Matched expected HTTP status code: 404
最後の応答の Code: 404

botocore.exceptions に定義されているクラスは、公式ドキュメントにある次のコードで一覧できます（この環境では 101 個でした）。

import botocore.exceptions

for key, value in sorted(botocore.exceptions.__dict__.items()):
    if isinstance(value, type):
        print(key)

まとめて受ける定型

import boto3
from botocore.exceptions import BotoCoreError, ClientError, ParamValidationError

s3 = boto3.client("s3")

def fetch(bucket, key):
    try:
        return s3.get_object(Bucket=bucket, Key=key)["Body"].read()
    except s3.exceptions.NoSuchKey:
        return None                                   # 想定内：無ければ None
    except ClientError as e:                          # AWS が返したエラー
        err, meta = e.response["Error"], e.response["ResponseMetadata"]
        print(f"AWS error op={e.operation_name} code={err['Code']} "
              f"http={meta['HTTPStatusCode']} request_id={meta['RequestId']}")
        raise
    except ParamValidationError as e:                 # 引数の誤り
        raise ValueError(f"引数を確認してください: {e}") from e
    except BotoCoreError as e:                        # 認証情報・接続など
        print(f"client-side error {type(e).__name__}: {e}")
        raise

print(fetch("rjc-demo-bucket", "report.csv"))

None

リトライ設定：legacy・standard・adaptiveの比較

boto3 は、一時的なエラーを自動で再送します。挙動は Config の retries で決めます。

import boto3
from botocore.config import Config

config = Config(retries={"mode": "standard", "total_max_attempts": 5})
sqs = boto3.client("sqs", config=config)

公式ドキュメント（Retries）の内容を表にします。

設定できる場所は 3 つです。

Config の max_attempts だけ意味が違うので、コードでは total_max_attempts を使うと、設定ファイルや環境変数と数え方がそろいます。

何回送られるかを数える

実際の AWS で 503 やスロットリングを狙って起こすことは難しいので、決まったエラー応答を返す小さな HTTP サーバーを立て、そこへ SQS の ListQueues を送って回数を数えます。次の 2 ファイルを同じフォルダーに置きます（このサーバーは回数を数えるためのもので、接続先を endpoint_url で切り替えています）。

# stub_server.py
"""決まった応答を返す検証用の HTTP サーバー（SQS の ListQueues 向け）"""
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

RESPONSES = {
    "503": (503, {"__type": "ServiceUnavailable", "message": "Service Unavailable"}),
    "throttle": (400, {"__type": "ThrottlingException", "message": "Rate exceeded"}),
    "denied": (403, {"__type": "AccessDeniedException", "message": "not authorized"}),
    "invalid": (400, {"__type": "InvalidParameterValue", "message": "invalid value"}),
}
state = {"mode": "503", "fail_times": 999, "requests": 0}

class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_POST(self):
        self.rfile.read(int(self.headers.get("Content-Length", 0)))
        state["requests"] += 1
        if state["requests"] <= state["fail_times"]:
            if state["mode"] == "slow":
                time.sleep(3)
                status, body = 200, {"QueueUrls": []}
            else:
                status, body = RESPONSES[state["mode"]]
        else:
            status, body = 200, {"QueueUrls": []}
        data = json.dumps(body).encode()
        try:
            self.send_response(status)
            self.send_header("Content-Type", "application/x-amz-json-1.0")
            self.send_header("x-amzn-RequestId", f"req-{state['requests']:04d}")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)
        except OSError:
            pass  # クライアントが先に切った場合

def start(port=9211):
    server = ThreadingHTTPServer(("127.0.0.1", port), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return f"http://127.0.0.1:{port}"

def reset(mode, fail_times=999):
    state.update(mode=mode, fail_times=fail_times, requests=0)

# retry_lab.py
import time
import boto3
from botocore.config import Config
from botocore.exceptions import BotoCoreError, ClientError
import stub_server

URL = stub_server.start()

def run(label, retries, mode, fail_times=999):
    stub_server.reset(mode, fail_times)
    sqs = boto3.client(
        "sqs", endpoint_url=URL, region_name="ap-northeast-1",
        aws_access_key_id="x", aws_secret_access_key="x",
        config=Config(retries=retries) if retries else None)
    started = time.time()
    try:
        meta = sqs.list_queues()["ResponseMetadata"]
        result = f"成功 RetryAttempts={meta['RetryAttempts']}"
    except ClientError as e:
        meta = e.response["ResponseMetadata"]
        result = f"{e.response['Error']['Code']} RetryAttempts={meta['RetryAttempts']}"
    except BotoCoreError as e:
        result = type(e).__name__
    print(f"{label:32} 送信 {stub_server.state['requests']} 回 {time.time() - started:4.1f} 秒  {result}")

print("== 503 を返し続ける")
run("指定なし", None, "503")
for mode in ["legacy", "standard", "adaptive"]:
    run(f"mode={mode}", {"mode": mode}, "503")
print("== ThrottlingException（HTTP 400）を返し続ける")
for mode in ["legacy", "standard", "adaptive"]:
    run(f"mode={mode}", {"mode": mode}, "throttle")
print("== リトライされないエラー")
for mode in ["legacy", "standard"]:
    run(f"mode={mode} AccessDenied", {"mode": mode}, "denied")
    run(f"mode={mode} InvalidParameter", {"mode": mode}, "invalid")
print("== max_attempts と total_max_attempts")
run("standard max_attempts=3", {"mode": "standard", "max_attempts": 3}, "503")
run("standard total_max_attempts=3", {"mode": "standard", "total_max_attempts": 3}, "503")
run("standard total_max_attempts=1", {"mode": "standard", "total_max_attempts": 1}, "503")
print("== 2 回エラーのあと成功する")
run("standard total_max_attempts=5", {"mode": "standard", "total_max_attempts": 5}, "503", fail_times=2)

== 503 を返し続ける
指定なし                             送信 5 回  8.7 秒  ServiceUnavailable RetryAttempts=4
mode=legacy                      送信 5 回  8.1 秒  ServiceUnavailable RetryAttempts=4
mode=standard                    送信 3 回  1.4 秒  ServiceUnavailable RetryAttempts=2
mode=adaptive                    送信 3 回  1.8 秒  ServiceUnavailable RetryAttempts=2
== ThrottlingException（HTTP 400）を返し続ける
mode=legacy                      送信 5 回  3.8 秒  ThrottlingException RetryAttempts=4
mode=standard                    送信 3 回  0.9 秒  ThrottlingException RetryAttempts=2
mode=adaptive                    送信 3 回  4.0 秒  ThrottlingException RetryAttempts=2
== リトライされないエラー
mode=legacy AccessDenied         送信 1 回  0.0 秒  AccessDeniedException RetryAttempts=0
mode=legacy InvalidParameter     送信 1 回  0.0 秒  InvalidParameterValue RetryAttempts=0
mode=standard AccessDenied       送信 1 回  0.0 秒  AccessDeniedException RetryAttempts=0
mode=standard InvalidParameter   送信 1 回  0.0 秒  InvalidParameterValue RetryAttempts=0
== max_attempts と total_max_attempts
standard max_attempts=3          送信 4 回  0.7 秒  ServiceUnavailable RetryAttempts=3
standard total_max_attempts=3    送信 3 回  1.6 秒  ServiceUnavailable RetryAttempts=2
standard total_max_attempts=1    送信 1 回  0.0 秒  ServiceUnavailable RetryAttempts=0
== 2 回エラーのあと成功する
standard total_max_attempts=5    送信 3 回  1.6 秒  成功 RetryAttempts=2

読み取れることを整理します。

かかった秒数は、待ち時間に乱数が入るため実行のたびに変わります。

リトライされるエラー・されないエラー

タイムアウトとリトライは掛け算になる

応答に 3 秒かかるサーバーへ、read_timeout=1 で送ります。

# timeout_demo.py
import time
import boto3
from botocore.config import Config
from botocore.exceptions import ReadTimeoutError
import stub_server

URL = stub_server.start()

for label, retries in [("total_max_attempts=3", 3), ("total_max_attempts=1", 1)]:
    stub_server.reset("slow")          # 応答まで 3 秒かかる
    sqs = boto3.client(
        "sqs", endpoint_url=URL, region_name="ap-northeast-1",
        aws_access_key_id="x", aws_secret_access_key="x",
        config=Config(connect_timeout=2, read_timeout=1,
                      retries={"mode": "standard", "total_max_attempts": retries}))
    started = time.time()
    try:
        sqs.list_queues()
    except ReadTimeoutError as e:
        print(f"{label}: 送信 {stub_server.state['requests']} 回 {time.time() - started:.1f} 秒")
        print("   ", type(e).__name__, ":", e)

total_max_attempts=3: 送信 3 回 4.0 秒
    ReadTimeoutError : Read timeout on endpoint URL: "http://127.0.0.1:9211/"
total_max_attempts=1: 送信 1 回 1.0 秒
    ReadTimeoutError : Read timeout on endpoint URL: "http://127.0.0.1:9211/"

1 回の呼び出しが返るまでの最長時間は、おおよそ「タイムアウト × 総回数 ＋ 待ち時間の合計」になります。read_timeout の既定は 60 秒なので、画面の応答や Lambda の実行時間に上限がある処理では、Config でタイムアウトと総回数の両方を決めておくと見積もりやすくなります。

自前バックオフとの二重リトライに注意

SDK の自動リトライの外側に自前のリトライを書くと、回数は掛け算になります。

# double_retry.py
import time
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
import stub_server

URL = stub_server.start()
RETRYABLE = {"ThrottlingException", "ServiceUnavailable"}

def make_client(**retries):
    return boto3.client(
        "sqs", endpoint_url=URL, region_name="ap-northeast-1",
        aws_access_key_id="x", aws_secret_access_key="x",
        config=Config(retries=retries))

def call_with_backoff(client, tries=3, base=0.2):
    """自前の指数バックオフ（リトライしてよいコードだけ）"""
    for attempt in range(1, tries + 1):
        try:
            return client.list_queues()
        except ClientError as e:
            if e.response["Error"]["Code"] not in RETRYABLE or attempt == tries:
                raise
            time.sleep(base * 2 ** (attempt - 1))

for label, client in [
    ("SDK 3 回 × 自前 3 回", make_client(mode="standard", total_max_attempts=3)),
    ("SDK 1 回 × 自前 3 回", make_client(mode="standard", total_max_attempts=1)),
]:
    stub_server.reset("throttle")
    started = time.time()
    try:
        call_with_backoff(client)
    except ClientError as e:
        print(f"{label}: 送信 {stub_server.state['requests']} 回 "
              f"{time.time() - started:.1f} 秒 -> {e.response['Error']['Code']}")

SDK 3 回 × 自前 3 回: 送信 9 回 4.1 秒 -> ThrottlingException
SDK 1 回 × 自前 3 回: 送信 3 回 0.6 秒 -> ThrottlingException

SDK が 3 回送るクライアントを自前のループで 3 回呼ぶと、合計 9 回送られました。スロットリング中のサービスに送る回数が増えると、回復までの時間も延びやすくなります。

DynamoDB の batch_write_item が返す UnprocessedItems のように、エラーにならずに「残り」が返る API は SDK の自動リトライの対象外です。公式ドキュメントは、残りを送り直すときに指数バックオフを使うことを強く勧めています。

よくある質問

Q. except Exception でまとめて受けてもよいですか？ A. 動きますが、原因の切り分けがしにくくなります。ClientError（AWS が返したエラー）と BotoCoreError（手元のエラー）を分けて受け、ClientError は Error.Code をログに残す形がおすすめです。

Q. エラーコードの文字列はどこで調べられますか？ A. 各サービスの API リファレンスのエラーの節です。手元では、一度エラーを起こして e.response["Error"]["Code"] を表示するのが確実です。クラス名とコードが違うサービス（IAM・SQS など）があります。

Q. リトライが本当に行われたかを確かめるには？ A. 応答（成功時も例外時も）の ResponseMetadata["RetryAttempts"] を見ます。公式ドキュメントによると、DEBUG ログにも再送の記録が出ます（legacy は botocore.retryhandler、standard・adaptive は botocore.retries.standard）。

Q. resource を使っているときは、どう捕まえますか？ A. resource.meta.client.exceptions.XXX か、client のときと同じ ClientError ＋ コード判定で捕まえます。

まとめ

• boto3 のエラーは ClientError（AWS が返した）と BotoCoreError（手元で起きた）の 2 系統で、親子関係はありません。
• ClientError は e.response["Error"]["Code"] で分岐し、RequestId・HTTPStatusCode・RetryAttempts をログに残します。
• client.exceptions.XXX は API 定義にあるエラーだけです。"404" や BucketNotEmpty のように、ClientError でしか受けられないコードがあります。
• クラス名とコードが違うサービスがあるので、コードは実際に出力して確かめます。
• リトライは指定なしだと legacy の 5 回、standard・adaptive は 3 回です。回数は total_max_attempts で指定すると数え方がそろいます。
• スロットリング・5xx・接続系は自動で再送され、権限・入力の誤り・対象なしは再送されません。
• 自前のバックオフを重ねると回数は掛け算になります。SDK に任せるか、SDK 側を 1 回にして自前で制御するかを決めます。

参考・出典

• Boto3 ドキュメント：Error handling
• Boto3 ドキュメント：Retries
• Amazon S3 API リファレンス：Error responses
• Amazon DynamoDB 開発者ガイド：Error handling with DynamoDB
• AWS STS API リファレンス：Common Errors

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