aws-sdk-python-usage

此技能启用时，请勿在任何代码、注释或输出中使用表情符号。

AWS SDK for Python（boto3）

boto3 是 AWS 的高级 Python SDK。它封装了 botocore（低级 SDK），并提供两种截然不同的接口：客户端（低级，与 API 一一对应）和资源（高级、面向对象）。了解何时使用哪种接口至关重要。

客户端与资源

客户端直接映射到 AWS 服务 API。每个服务都有一个客户端。响应是普通字典。

资源提供了带有属性和操作的面向对象接口。只有部分服务提供资源（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 的大小写完全一致，通常采用 PascalCase，而非 snake_case。

错误处理

仅在有可采取的实际操作时捕获异常，例如返回备用值、重试或采用其他代码路径。仅仅为了打印并吞掉异常而捕获它是不对的：这样会掩盖真正的错误，并阻止调用方作出响应。默认情况下，应让异常继续传播。

确实需要捕获时，优先使用客户端上的类型化异常，而不是通过 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 仅应在顶层错误处理程序中用作兜底，不要在业务逻辑函数中使用。它属于 botocore，而非 boto3：

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__" 中只包含一次函数调用。参数解析、错误展示和退出码应放在 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