aws-sdk-python-usage
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseDo not use emojis in any code, comments, or output when this skill is active.
当本技能生效时,请勿在任何代码、注释或输出中使用表情符号。
AWS SDK for Python (boto3)
AWS SDK for Python (boto3)
boto3 is the high-level Python SDK for AWS. It wraps botocore (the low-level
SDK) and provides two distinct interfaces: clients (low-level, 1:1 API
mapping) and resources (high-level, object-oriented). Understanding which to
use and when is essential.
boto3是AWS的高级Python SDK。它封装了botocore(低级SDK)并提供两种截然不同的接口:clients(低级,与API一对一映射)和resources(高级,面向对象)。理解何时使用哪种接口至关重要。
Client vs Resource
Client 与 Resource对比
Clients map directly to AWS service APIs. Every service has a client.
Responses are plain dicts.
Resources provide an object-oriented interface with attributes and actions.
Only some services have resources (S3, DynamoDB, EC2, IAM, SQS, SNS,
CloudFormation, CloudWatch, Glacier). Resources auto-marshal types (especially
useful for DynamoDB).
python
import boto3Clients直接映射到AWS服务API。每个服务都有对应的client。返回结果为普通字典。
Resources提供带有属性和操作的面向对象接口。仅部分服务支持resources(S3、DynamoDB、EC2、IAM、SQS、SNS、CloudFormation、CloudWatch、Glacier)。Resources会自动转换类型(对DynamoDB尤其有用)。
python
import boto3Client - low-level, all services
Client - 低级,支持所有服务
s3_client = boto3.client("s3")
response = s3_client.list_buckets()
buckets = response["Buckets"] # plain dicts
s3_client = boto3.client("s3")
response = s3_client.list_buckets()
buckets = response["Buckets"] # 普通字典
Resource - high-level, select services
Resource - 高级,仅支持部分服务
s3_resource = boto3.resource("s3")
for bucket in s3_resource.buckets.all():
print(bucket.name) # attribute access, not dict keys
Use clients when you need full API coverage or the service has no resource
interface. Use resources when they exist and simplify your code (especially
DynamoDB and S3).s3_resource = boto3.resource("s3")
for bucket in s3_resource.buckets.all():
print(bucket.name) # 属性访问,而非字典键
当您需要完整的API覆盖,或该服务没有resource接口时,请使用clients。当resource接口存在且能简化代码时(尤其是DynamoDB和S3),请使用resources。Session and Client Creation
会话与客户端创建
python
import boto3python
import boto3Default session implicitly created
隐式创建默认会话
client = boto3.client("s3")
resource = boto3.resource("dynamodb")
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")
Do not create clients inside loops - reuse a single client instance. Clients
are thread safe and can be shared across threads once they're instantiated.session = boto3.Session(
profile_name="my-profile",
region_name="us-west-2",
)
client = session.client("s3")
请勿在循环内创建客户端——复用单个客户端实例。客户端是线程安全的,实例化后可在多线程间共享。Making API Calls
调用API
python
undefinedpython
undefinedClient - pass parameters as keyword arguments, get dicts back
Client - 以关键字参数传递参数,返回字典
response = client.get_object(Bucket="my-bucket", Key="my-key")
data = response["Body"].read()
response = client.get_object(Bucket="my-bucket", Key="my-key")
data = response["Body"].read()
Resource - use object methods and attributes
Resource - 使用对象方法和属性
obj = s3_resource.Object("my-bucket", "my-key")
response = obj.get()
data = response["Body"].read()
Parameter names match the exact casing of the AWS API,
which is typically PascalCase, not snake\_case.obj = s3_resource.Object("my-bucket", "my-key")
response = obj.get()
data = response["Body"].read()
参数名称与AWS API的大小写完全匹配,通常为PascalCase,而非snake_case。Error Handling
错误处理
Only catch exceptions when you have something actionable to do - return a
fallback value, retry, take a different code path. Catching an exception just to
print it and swallow it is wrong: it hides the real error and prevents callers
from reacting. Let exceptions propagate by default.
When you do catch, prefer typed exceptions on the client over generic
with string code matching through the
attribute:
ClientErrorclient.exceptionspython
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 itUse generic only as a catch-all in a top-level error handler, not
in business logic functions. It lives in botocore, not boto3:
ClientErrorpython
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 1For the full error hierarchy and botocore exceptions, see .
references/error-handling.md仅当您有可执行的操作时才捕获异常——返回回退值、重试、选择不同的代码路径。仅捕获异常来打印并忽略是错误的:这会隐藏真实错误并阻止调用者做出响应。默认情况下让异常向上传播。
当您确实需要捕获异常时,优先使用客户端上的类型化异常,而非通过属性匹配字符串代码的通用:
client.exceptionsClientErrorpython
lambda_client = boto3.client("lambda")
def get_function_config(name: str) -> dict | None:
"""返回函数配置,如果函数不存在则返回None。"""
try:
return lambda_client.get_function_configuration(FunctionName=name)
except lambda_client.exceptions.ResourceNotFoundException:
return None # 可执行操作:将不存在的函数转换为None
# 其他所有异常向上传播——由调用者或main()处理仅在顶层错误处理程序中使用通用作为捕获所有异常的手段,不要在业务逻辑函数中使用。它属于botocore,而非boto3:
ClientErrorpython
from botocore.exceptions import ClientError
def main() -> int:
try:
result = do_the_work()
print(result)
return 0
except ClientError as e:
print(f"错误: {e}", file=sys.stderr)
return 1关于完整的错误层级和botocore异常,请参阅。
references/error-handling.mdScript Structure
脚本结构
When asked to write a script that uses or , keep to a single function call. Argument parsing, error presentation,
and exit codes belong in , not scattered across business logic
functions:
boto3botocoreif __name__ == "__main__"main()python
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())Never call from a business logic function -- it makes the function
untestable and unusable as a library. Raise an exception or return an error
value instead, and let decide how to present it.
sys.exit()main()当要求编写使用或的脚本时,将仅用于单个函数调用。参数解析、错误展示和退出码应放在中,不要分散在业务逻辑函数中:
boto3botocoreif __name__ == "__main__"main()python
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"错误: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())切勿从业务逻辑函数中调用——这会使函数无法测试且无法作为库使用。请改为抛出异常或返回错误值,由决定如何展示。
sys.exit()main()Pagination
分页
Never manually loop with -- use paginators. When you only need
specific fields, use with a JMESPath expression to extract and
flatten across pages:
NextToken.search()python
paginator = iam.get_paginator("list_users")
for name in paginator.paginate().search("Users[].UserName"):
print(name)切勿手动使用循环——请使用分页器。当您仅需要特定字段时,使用配合JMESPath表达式来提取并跨页展平数据:
NextToken.search()python
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)
When you need the full response object per item, or need per-page control (e.g.
counting pages, batching by page), iterate pages directly:
```python
for page in paginator.paginate():
for user in page.get("Users", []):
process(user)For more details on pagination, see: .
references/pagination.mdfor arn in paginator.paginate().search("Users[?Path == '/admin/'][].Arn"):
print(arn)
当您需要每个项的完整响应对象,或需要每页控制(例如统计页数、按页分批处理)时,请直接迭代页面:
```python
for page in paginator.paginate():
for user in page.get("Users", []):
process(user)有关分页的更多详细信息,请参阅:。
references/pagination.mdWaiters
等待器
Wait for a resource to reach a desired state:
python
waiter = client.get_waiter("bucket_exists")
waiter.wait(
Bucket="my-bucket",
WaiterConfig={"Delay": 5, "MaxAttempts": 20},
)For more details on waiters, see .
references/waiters.md等待资源达到期望状态:
python
waiter = client.get_waiter("bucket_exists")
waiter.wait(
Bucket="my-bucket",
WaiterConfig={"Delay": 5, "MaxAttempts": 20},
)有关等待器的更多详细信息,请参阅。
references/waiters.mdClient Configuration
客户端配置
Use for retries, timeouts, and connection pool
settings, etc.:
botocore.config.Configpython
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)When creating custom configuration for a client, see .
references/configuration.md使用配置重试、超时和连接池设置等:
botocore.config.Configpython
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.mdLogging
日志记录
Both boto3 and botocore use the standard library module. You can
configure logging through the standard APIs, or you can use
helpers provided by boto3 and botocore for convenience:
loggingloggingpython
undefinedboto3和botocore均使用标准库模块。您可以通过标准 API配置日志记录,也可以使用boto3和botocore提供的辅助工具以简化操作:
loggingloggingpython
undefinedQuick: log all botocore wire-level details to stderr
快速配置:将所有botocore网络级细节记录到stderr
boto3.set_stream_logger("") # root logger -- everything
boto3.set_stream_logger("botocore") # just botocore
boto3.set_stream_logger("") # 根日志记录器——所有内容
boto3.set_stream_logger("botocore") # 仅记录botocore
Botocore, log all botocore details
Botocore:记录所有botocore细节
import logging
from botocore.session import Session
session = Session()
session.set_stream_logger('botocore', logging.DEBUG)
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)` adds a
`StreamHandler` to the named logger. This is the idiomatic way to get
request/response debug output from the SDK.session.set_file_logger(logging.DEBUG, '/tmp/botocore.log')
`set_stream_logger(name, level=logging.DEBUG)`会为指定的日志记录器添加一个`StreamHandler`。这是从SDK获取请求/响应调试输出的惯用方式。Common Issues
常见问题
Issue: ClientError import location
问题:ClientError导入位置
Wrong:
Right:
from boto3.exceptions import ClientErrorfrom botocore.exceptions import ClientError错误写法:
正确写法:
from boto3.exceptions import ClientErrorfrom botocore.exceptions import ClientErrorService specific customizations
特定服务自定义
When writing any Python code that uses the following services, you MUST load
these additional reference files for best practices and custom high level APIs:
- S3 - you MUST load .
references/s3.md - Dynamodb - you MUST load .
references/dynamodb.md
当编写使用以下服务的Python代码时,您必须加载这些额外的参考文件以遵循最佳实践并使用自定义高级API:
- S3 - 您必须加载。
references/s3.md - Dynamodb - 您必须加载。
references/dynamodb.md
References
参考资料
- Client configuration (retries, timeouts, endpoints):
references/configuration.md - Credentials and sessions:
references/credentials.md - Error handling patterns:
references/error-handling.md - Pagination:
references/pagination.md - Waiters:
references/waiters.md - S3 transfers and presigned URLs:
references/s3.md - DynamoDB operations:
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