aws-sdk-python-usage

このスキルが有効な間は、コード、コメント、出力に絵文字を使用しないでください。

AWS SDK for Python（boto3）

boto3 は AWS 用の高レベルな Python SDK です。低レベル SDK である botocore をラップし、クライアント（低レベルで、API と 1 対 1 に対応）とリソース（高レベルで、オブジェクト指向）の 2 つの異なるインターフェースを提供します。どちらをいつ使うかを理解することが重要です。

クライアントとリソース

クライアントは AWS サービスの API に直接対応します。すべてのサービスにクライアントがあります。レスポンスは通常の dict です。

リソースは、属性とアクションを備えたオブジェクト指向のインターフェースを提供します。リソースを利用できるサービスは限られています（S3、DynamoDB、EC2、IAM、SQS、SNS、CloudFormation、CloudWatch、Glacier）。リソースは型を自動的にマーシャリングします（特に DynamoDB で便利です）。

import boto3

# Client - low-level, all services
s3_client = boto3.client("s3")
response = s3_client.list_buckets()
buckets = response["Buckets"]  # plain dicts

# Resource - high-level, select services
s3_resource = boto3.resource("s3")
for bucket in s3_resource.buckets.all():
    print(bucket.name)  # attribute access, not dict keys

API をすべて利用する必要がある場合や、サービスにリソースインターフェースがない場合は、クライアントを使用してください。リソースが利用でき、コードが簡潔になる場合（特に DynamoDB や S3）は、リソースを使用してください。

セッションとクライアントの作成

import boto3

# Default session implicitly created
client = boto3.client("s3")
resource = boto3.resource("dynamodb")

# Explicit session use when you need to customize how
# clients are created, use an explicit profile, etc.
session = boto3.Session(
    profile_name="my-profile",
    region_name="us-west-2",
)
client = session.client("s3")

ループ内でクライアントを作成しないでください。単一のクライアントインスタンスを再利用してください。クライアントはスレッドセーフであり、インスタンス化後はスレッド間で共有できます。

API 呼び出し

# Client - pass parameters as keyword arguments, get dicts back
response = client.get_object(Bucket="my-bucket", Key="my-key")
data = response["Body"].read()

# Resource - use object methods and attributes
obj = s3_resource.Object("my-bucket", "my-key")
response = obj.get()
data = response["Body"].read()

パラメーター名は AWS API の大文字と小文字の区別に正確に一致します。通常は snake_case ではなく PascalCase です。

エラー処理

実行可能な対処（代替値を返す、再試行する、別のコードパスを実行する）がある場合にのみ、例外を捕捉してください。例外を捕捉して表示するだけで握りつぶすのは誤りです。実際のエラーが隠され、呼び出し元が対処できなくなります。通常は例外をそのまま伝播させてください。

例外を捕捉する場合は、client.exceptions 属性を通じて文字列コードを照合する汎用の ClientError よりも、クライアント上の型付き例外を優先してください。

lambda_client = boto3.client("lambda")

def get_function_config(name: str) -> dict | None:
    """Return function configuration, or None if it doesn't exist."""
    try:
        return lambda_client.get_function_configuration(FunctionName=name)
    except lambda_client.exceptions.ResourceNotFoundException:
        return None  # actionable: convert missing function to None
    # Everything else propagates - caller or main() handles it

汎用の ClientError は、最上位のエラーハンドラーで包括的に捕捉する場合にのみ使用し、ビジネスロジック関数では使用しないでください。これは boto3 ではなく botocore にあります。

from botocore.exceptions import ClientError

def main() -> int:
    try:
        result = do_the_work()
        print(result)
        return 0
    except ClientError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1

エラーの階層全体と botocore の例外については、references/error-handling.md を参照してください。

スクリプトの構成

boto3 または botocore を使用するスクリプトの作成を依頼された場合は、if __name__ == "__main__" の中身を 1 つの関数呼び出しだけにしてください。引数の解析、エラーの表示、終了コードの処理は main() に記述し、ビジネスロジック関数に分散させないでください。

def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("bucket")
    args = parser.parse_args()

    try:
        do_the_work(args.bucket)
        return 0
    except ClientError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1

if __name__ == "__main__":
    sys.exit(main())

ビジネスロジック関数から sys.exit() を呼び出さないでください。関数がテスト不能になり、ライブラリとしても使用できなくなります。代わりに例外を送出するかエラー値を返し、どのように表示するかは main() に判断させてください。

ページネーション

NextToken を使って手動でループしないでください。ページネーターを使用してください。特定のフィールドだけが必要な場合は、.search() と JMESPath 式を使って、ページをまたいで抽出し、フラット化してください。

paginator = iam.get_paginator("list_users")
for name in paginator.paginate().search("Users[].UserName"):
    print(name)

# Filter and project
for arn in paginator.paginate().search("Users[?Path == '/admin/'][].Arn"):
    print(arn)

各項目のレスポンスオブジェクト全体が必要な場合や、ページ単位の制御（ページ数のカウント、ページごとのバッチ処理など）が必要な場合は、ページを直接反復処理してください。

for page in paginator.paginate():
    for user in page.get("Users", []):
        process(user)

ページネーションの詳細については、references/pagination.md を参照してください。

ウェイター

リソースが目的の状態になるまで待機します。

waiter = client.get_waiter("bucket_exists")
waiter.wait(
    Bucket="my-bucket",
    WaiterConfig={"Delay": 5, "MaxAttempts": 20},
)

ウェイターの詳細については、references/waiters.md を参照してください。

クライアントの設定

再試行、タイムアウト、接続プールの設定などには、botocore.config.Config を使用してください。

from botocore.config import Config

config = Config(
    retries={"total_max_attempts": 2, "mode": "adaptive"},
    connect_timeout=5,
    read_timeout=10,
    max_pool_connections=50,
)
client = boto3.client("s3", config=config)

クライアントにカスタム設定を作成する場合は、references/configuration.md を参照してください。

ロギング

boto3 と botocore はどちらも、標準ライブラリの logging モジュールを使用します。標準の logging API でロギングを設定することも、便利な boto3 および botocore のヘルパーを使用することもできます。

# Quick: log all botocore wire-level details to stderr
boto3.set_stream_logger("")  # root logger -- everything
boto3.set_stream_logger("botocore")  # just botocore

# Botocore, log all botocore details
import logging

from botocore.session import Session

session = Session()

session.set_stream_logger('botocore', logging.DEBUG)
# OR: Configure logging to a file.
session.set_file_logger(logging.DEBUG, '/tmp/botocore.log')

set_stream_logger(name, level=logging.DEBUG) は、指定したロガーに StreamHandler を追加します。SDK のリクエストとレスポンスのデバッグ出力を取得する標準的な方法です。

よくある問題

問題：ClientError のインポート元

誤り： from boto3.exceptions import ClientError 正しい： from botocore.exceptions import ClientError

サービス固有のカスタマイズ

以下のサービスを使用する Python コードを記述する場合は、ベストプラクティスと高レベル API に関する追加のリファレンスファイルを必ず読み込んでください。

• S3 - references/s3.md を必ず読み込んでください。
• Dynamodb - references/dynamodb.md を必ず読み込んでください。

リファレンス

• クライアントの設定（再試行、タイムアウト、エンドポイント）：references/configuration.md
• 認証情報とセッション：references/credentials.md
• エラーハンドリングのパターン：references/error-handling.md
• ページネーション：references/pagination.md
• ウェイター：references/waiters.md
• S3 の転送と署名付き URL：references/s3.md
• DynamoDB の操作：references/dynamodb.md