chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,260 @@
|
||||
---
|
||||
name: aws-sdk-python-usage
|
||||
description: |
|
||||
AWS SDK for Python (boto3/botocore) development patterns. You MUST use this skill when writing Python code that uses AWS services via boto3 or botocore. This includes creating service clients or resources, configuring sessions and credentials, handling errors with ClientError, using paginators and waiters, S3 file transfers and presigned URLs, DynamoDB table operations, and any boto3/botocore client configuration. Use this skill whenever Python code imports boto3 or botocore, or when the user asks about AWS operations in Python.
|
||||
---
|
||||
|
||||
> Do not use emojis in any code, comments, or output when this skill is active.
|
||||
|
||||
# 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.
|
||||
|
||||
## Client vs 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 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
|
||||
```
|
||||
|
||||
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).
|
||||
|
||||
## Session and Client Creation
|
||||
|
||||
```python
|
||||
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")
|
||||
```
|
||||
|
||||
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.
|
||||
|
||||
## Making API Calls
|
||||
|
||||
```python
|
||||
# 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()
|
||||
```
|
||||
|
||||
Parameter names match the exact casing of the AWS API,
|
||||
which is typically PascalCase, not 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
|
||||
`ClientError` with string code matching through the `client.exceptions`
|
||||
attribute:
|
||||
|
||||
```python
|
||||
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
|
||||
```
|
||||
|
||||
Use generic `ClientError` only as a catch-all in a top-level error handler, not
|
||||
in business logic functions. It lives in botocore, not boto3:
|
||||
|
||||
```python
|
||||
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
|
||||
```
|
||||
|
||||
For the full error hierarchy and botocore exceptions, see `references/error-handling.md`.
|
||||
|
||||
## Script Structure
|
||||
|
||||
When asked to write a script that uses `boto3` or `botocore`, keep `if __name__
|
||||
== "__main__"` to a single function call. Argument parsing, error presentation,
|
||||
and exit codes belong in `main()`, not scattered across business logic
|
||||
functions:
|
||||
|
||||
```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 `sys.exit()` 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 `main()` decide how to present it.
|
||||
|
||||
## Pagination
|
||||
|
||||
Never manually loop with `NextToken` -- use paginators. When you only need
|
||||
specific fields, use `.search()` with a JMESPath expression to extract and
|
||||
flatten across pages:
|
||||
|
||||
```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.md`.
|
||||
|
||||
## Waiters
|
||||
|
||||
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`.
|
||||
|
||||
## Client Configuration
|
||||
|
||||
Use `botocore.config.Config` for retries, timeouts, and connection pool
|
||||
settings, etc.:
|
||||
|
||||
```python
|
||||
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`.
|
||||
|
||||
## Logging
|
||||
|
||||
Both boto3 and botocore use the standard library `logging` module. You can
|
||||
configure logging through the standard `logging` APIs, or you can use
|
||||
helpers provided by boto3 and botocore for convenience:
|
||||
|
||||
```python
|
||||
# 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)` adds a
|
||||
`StreamHandler` to the named logger. This is the idiomatic way to get
|
||||
request/response debug output from the SDK.
|
||||
|
||||
## Common Issues
|
||||
|
||||
### Issue: ClientError import location
|
||||
|
||||
**Wrong:** `from boto3.exceptions import ClientError`
|
||||
**Right:** `from botocore.exceptions import ClientError`
|
||||
|
||||
## Service 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`.
|
||||
|
||||
## 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`
|
||||
@@ -0,0 +1,151 @@
|
||||
# Client Configuration Reference
|
||||
|
||||
## botocore.config.Config
|
||||
|
||||
All configuration is passed via `botocore.config.Config`. Multiple configs can be merged:
|
||||
|
||||
```python
|
||||
from botocore.config import Config
|
||||
|
||||
base = Config(retries={"total_max_attempts": 2, "mode": "standard"})
|
||||
s3_specific = Config(s3={"addressing_style": "path"})
|
||||
|
||||
# Merge -- later config wins on conflicts
|
||||
client = boto3.client("s3", config=base.merge(s3_specific))
|
||||
```
|
||||
|
||||
## Retry Configuration
|
||||
|
||||
```python
|
||||
config = Config(
|
||||
retries={
|
||||
"total_max_attempts": 2, # total attempts including first try (1 retry attempt here)
|
||||
"mode": "adaptive", # legacy | standard | adaptive
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
Prefer using `total_max_attempts` over the legacy `max_attempts`. The
|
||||
`max_attempts` value does not include the first attempt (it's actually the
|
||||
number of retry attempts).
|
||||
|
||||
| Mode | Default attempts | Behavior |
|
||||
|---|---|---|
|
||||
| `legacy` | 5 | Retries on a limited set of errors |
|
||||
| `standard` | 3 | Broader retryable errors, consistent exponential backoff |
|
||||
| `adaptive` | 3 | Standard + client-side rate limiting (token bucket) |
|
||||
|
||||
Can also set via `AWS_MAX_ATTEMPTS` and `AWS_RETRY_MODE` env vars or `~/.aws/config`.
|
||||
|
||||
## Timeouts
|
||||
|
||||
```python
|
||||
config = Config(
|
||||
connect_timeout=5, # seconds to establish connection (default 60)
|
||||
read_timeout=10, # seconds to wait for response data (default 60)
|
||||
)
|
||||
```
|
||||
|
||||
## Connection Pool
|
||||
|
||||
```python
|
||||
config = Config(
|
||||
max_pool_connections=50, # default 10 per client
|
||||
)
|
||||
```
|
||||
|
||||
Each client maintains its own urllib3 connection pool. If you're making parallel requests (e.g. with `concurrent.futures`), set `max_pool_connections` to match your concurrency level to avoid connection churn.
|
||||
|
||||
## Custom Endpoints
|
||||
|
||||
```python
|
||||
# Custom S3 endpoint on localhost.
|
||||
client = boto3.client(
|
||||
"s3",
|
||||
endpoint_url="http://localhost:4566",
|
||||
region_name="us-east-1",
|
||||
)
|
||||
|
||||
# FIPS endpoints
|
||||
config = Config(use_fips_endpoint=True)
|
||||
client = boto3.client("s3", config=config)
|
||||
|
||||
# Dual-stack (IPv4 + IPv6)
|
||||
config = Config(use_dualstack_endpoint=True)
|
||||
client = boto3.client("s3", config=config)
|
||||
```
|
||||
|
||||
## Proxy Configuration
|
||||
|
||||
```python
|
||||
# Via environment variables (preferred)
|
||||
# HTTP_PROXY=http://proxy:8080
|
||||
# HTTPS_PROXY=http://proxy:8080
|
||||
|
||||
# Via Config
|
||||
config = Config(
|
||||
proxies={"https": "http://proxy:8080"},
|
||||
proxies_config={"proxy_ca_bundle": "/path/to/ca-bundle.crt"},
|
||||
)
|
||||
```
|
||||
|
||||
## S3-Specific Configuration
|
||||
|
||||
```python
|
||||
config = Config(
|
||||
s3={
|
||||
"addressing_style": "path", # path | virtual | auto (default)
|
||||
"payload_signing_enabled": False, # skip payload signing for large uploads
|
||||
"us_east_1_regional_endpoint": "regional",
|
||||
},
|
||||
signature_version="s3v4",
|
||||
)
|
||||
|
||||
# Transfer acceleration
|
||||
config = Config(s3={"use_accelerate_endpoint": True})
|
||||
```
|
||||
|
||||
## User-Agent Customization
|
||||
|
||||
```python
|
||||
config = Config(
|
||||
user_agent_appid="my-app/1.0",
|
||||
user_agent_extra="custom-metadata",
|
||||
)
|
||||
```
|
||||
|
||||
## Sharing Config Across Clients
|
||||
|
||||
```python
|
||||
from botocore.config import Config
|
||||
|
||||
config = Config(
|
||||
retries={"total_max_attempts": 2, "mode": "standard"},
|
||||
connect_timeout=5,
|
||||
read_timeout=10,
|
||||
)
|
||||
|
||||
# Same config for multiple clients
|
||||
s3 = boto3.client("s3", config=config)
|
||||
dynamodb = boto3.client("dynamodb", config=config)
|
||||
lambda_client = boto3.client("lambda", config=config)
|
||||
```
|
||||
|
||||
You can also set a default client config in a botocore Session:
|
||||
|
||||
```python
|
||||
from botocore.config import Config
|
||||
from botocore.session import Session
|
||||
|
||||
config = Config(
|
||||
retries={"total_max_attempts": 2, "mode": "standard"},
|
||||
connect_timeout=5,
|
||||
read_timeout=10,
|
||||
)
|
||||
session = Session()
|
||||
session.set_default_client_config(config)
|
||||
# Now all clients created will use this session-specific default
|
||||
# config if an explicit config is not provided.
|
||||
s3 = session.create_client('s3')
|
||||
dynamodb = session.create_client('dynamodb')
|
||||
```
|
||||
@@ -0,0 +1,127 @@
|
||||
# Credentials Reference
|
||||
|
||||
## Default Credential Chain
|
||||
|
||||
boto3 resolves credentials in this order:
|
||||
|
||||
1. Explicit `aws_access_key_id`/`aws_secret_access_key` passed to `Session()` or `client()`
|
||||
2. `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` / `AWS_SESSION_TOKEN` env vars
|
||||
3. Assume role (`role_arn` + `source_profile` / `credential_source` in the active profile)
|
||||
4. Web identity token (EKS IRSA via `AWS_WEB_IDENTITY_TOKEN_FILE` / `AWS_ROLE_ARN`, or `web_identity_token_file` in profile)
|
||||
5. SSO credentials (IAM Identity Center profile; token from `aws sso login`)
|
||||
6. `~/.aws/credentials` file (default or named profile)
|
||||
7. Login session (`login_session` in profile; requires `botocore[crt]`)
|
||||
8. Credential process (`credential_process` in profile)
|
||||
9. `~/.aws/config` file (static keys in profile)
|
||||
10. Legacy boto config (`BOTO_CONFIG`, `~/.boto`, `/etc/boto.cfg`)
|
||||
11. Container credentials — ECS task role / EKS Pod Identity (`AWS_CONTAINER_CREDENTIALS_RELATIVE_URI` or `AWS_CONTAINER_CREDENTIALS_FULL_URI`)
|
||||
12. EC2 instance metadata (IMDS)
|
||||
|
||||
In most cases, let the default chain handle credential resolution rather than hardcoding credentials.
|
||||
|
||||
## Sessions
|
||||
|
||||
```python
|
||||
import boto3
|
||||
|
||||
# Default session -- shared across boto3.client()/boto3.resource() calls
|
||||
client = boto3.client("s3")
|
||||
|
||||
# Explicit session -- isolated credentials and config
|
||||
session = boto3.Session(
|
||||
profile_name="dev-account",
|
||||
region_name="us-west-2",
|
||||
)
|
||||
client = session.client("s3")
|
||||
|
||||
# Multiple sessions for cross-account access
|
||||
dev = boto3.Session(profile_name="dev")
|
||||
prod = boto3.Session(profile_name="prod")
|
||||
dev_s3 = dev.client("s3")
|
||||
prod_s3 = prod.client("s3")
|
||||
```
|
||||
|
||||
Use explicit sessions when you need multiple credential sets or profiles in the same process.
|
||||
|
||||
## Named Profiles
|
||||
|
||||
```python
|
||||
# Use a profile from ~/.aws/credentials or ~/.aws/config
|
||||
session = boto3.Session(profile_name="my-profile")
|
||||
client = session.client("s3")
|
||||
|
||||
# Or set via environment variable
|
||||
# AWS_PROFILE=my-profile
|
||||
```
|
||||
|
||||
## Assume Role (STS)
|
||||
|
||||
```python
|
||||
import boto3
|
||||
|
||||
# Assume a role and create a client with the temporary credentials
|
||||
sts = boto3.client("sts")
|
||||
response = sts.assume_role(
|
||||
RoleArn="arn:aws:iam::123456789012:role/MyRole",
|
||||
RoleSessionName="my-session",
|
||||
DurationSeconds=3600,
|
||||
)
|
||||
creds = response["Credentials"]
|
||||
|
||||
client = boto3.client(
|
||||
"s3",
|
||||
aws_access_key_id=creds["AccessKeyId"],
|
||||
aws_secret_access_key=creds["SecretAccessKey"],
|
||||
aws_session_token=creds["SessionToken"],
|
||||
)
|
||||
```
|
||||
|
||||
For automatic credential refresh when the assumed role expires, use a profile with `role_arn` in `~/.aws/config`:
|
||||
|
||||
```ini
|
||||
[profile cross-account]
|
||||
role_arn = arn:aws:iam::123456789012:role/MyRole
|
||||
source_profile = default
|
||||
```
|
||||
|
||||
```python
|
||||
session = boto3.Session(profile_name="cross-account")
|
||||
client = session.client("s3") # credentials auto-refresh
|
||||
```
|
||||
|
||||
## Chained Role Assumption
|
||||
|
||||
```ini
|
||||
# ~/.aws/config
|
||||
[profile role-a]
|
||||
role_arn = arn:aws:iam::111111111111:role/RoleA
|
||||
source_profile = default
|
||||
|
||||
[profile role-b]
|
||||
role_arn = arn:aws:iam::222222222222:role/RoleB
|
||||
source_profile = role-a
|
||||
```
|
||||
|
||||
## Environment Variables
|
||||
|
||||
| Variable | Purpose |
|
||||
|---|---|
|
||||
| `AWS_ACCESS_KEY_ID` | Access key |
|
||||
| `AWS_SECRET_ACCESS_KEY` | Secret key |
|
||||
| `AWS_SESSION_TOKEN` | Session token (temporary creds) |
|
||||
| `AWS_DEFAULT_REGION` | Default region |
|
||||
| `AWS_PROFILE` | Named profile |
|
||||
| `AWS_ROLE_ARN` | Role ARN for web identity |
|
||||
| `AWS_WEB_IDENTITY_TOKEN_FILE` | Path to OIDC token file (EKS) |
|
||||
| `AWS_CONFIG_FILE` | Override config file path |
|
||||
| `AWS_SHARED_CREDENTIALS_FILE` | Override credentials file path |
|
||||
|
||||
## STS Get Caller Identity
|
||||
|
||||
Useful for verifying which credentials are in use:
|
||||
|
||||
```python
|
||||
sts = boto3.client("sts")
|
||||
identity = sts.get_caller_identity()
|
||||
print(identity["Account"], identity["Arn"])
|
||||
```
|
||||
@@ -0,0 +1,295 @@
|
||||
# DynamoDB Reference
|
||||
|
||||
Use the resource interface to work with native Python types instead of AttributeValue dicts:
|
||||
|
||||
```python
|
||||
import boto3
|
||||
from boto3.dynamodb.conditions import Key, Attr
|
||||
|
||||
table = boto3.resource("dynamodb").Table("my-table")
|
||||
table.put_item(Item={"pk": "user#1", "name": "Alice", "age": 30})
|
||||
item = table.get_item(Key={"pk": "user#1"}).get("Item")
|
||||
```
|
||||
|
||||
## Common Pitfall: AttributeValue Dicts
|
||||
|
||||
If you see `{"id": {"S": "1"}, "count": {"N": "42"}}` instead of `{"id": "1", "count": 42}`, you're using `boto3.client("dynamodb")` which does not auto-marshal types. You have two options:
|
||||
|
||||
1. Use the **resource interface** (recommended) -- `Table` methods auto-marshal types.
|
||||
2. Use the **resource's underlying client** -- a low-level client that still
|
||||
auto-marshals types is available through the `.meta.client` attribute of a
|
||||
resource type:
|
||||
|
||||
```python
|
||||
|
||||
# Instead of: boto3.client('dynamodb')
|
||||
# you can use `boto3.resource('dynamodb').meta.client`.
|
||||
# This is still a boto3 DynamoDB client with custom handlers
|
||||
# to automatically marshal to the AttributeValue dict types.
|
||||
dynamodb = boto3.resource("dynamodb").meta.client
|
||||
# This client auto-converts Python types to/from DynamoDB AttributeValue format
|
||||
response = dynamodb.get_item(TableName="my-table", Key={"pk": "user#1"})
|
||||
item = response.get("Item") # {"pk": "user#1", "name": "Alice"} -- plain Python types
|
||||
```
|
||||
|
||||
ALWAYS prefer using native python types instead of low level AttributeValue
|
||||
dicts. These are more idiomatic for Python developers to work with and handle the
|
||||
conversion and various edge cases automatically for you.
|
||||
|
||||
## Error Handling
|
||||
|
||||
Access typed exceptions via `table.meta.client.exceptions` (not directly on the table):
|
||||
|
||||
```python
|
||||
table = boto3.resource("dynamodb").Table("my-table")
|
||||
|
||||
try:
|
||||
table.put_item(
|
||||
Item=new_item,
|
||||
ConditionExpression=Attr("pk").not_exists(),
|
||||
)
|
||||
except table.meta.client.exceptions.ConditionalCheckFailedException:
|
||||
# Actionable: item was created by another process, re-fetch it
|
||||
return table.get_item(Key={"pk": new_item["pk"]})["Item"]
|
||||
```
|
||||
|
||||
## Resource Interface (Recommended)
|
||||
|
||||
The resource interface automatically marshals between Python types and DynamoDB's type system:
|
||||
|
||||
```python
|
||||
import boto3
|
||||
from boto3.dynamodb.conditions import Key, Attr
|
||||
from decimal import Decimal
|
||||
|
||||
table = boto3.resource("dynamodb").Table("my-table")
|
||||
```
|
||||
|
||||
### CRUD Operations
|
||||
|
||||
```python
|
||||
# Put item
|
||||
table.put_item(Item={"pk": "user#1", "sk": "profile", "name": "Alice", "age": 30})
|
||||
|
||||
# Get item
|
||||
response = table.get_item(Key={"pk": "user#1", "sk": "profile"})
|
||||
item = response.get("Item") # None if not found
|
||||
|
||||
# Update item
|
||||
table.update_item(
|
||||
Key={"pk": "user#1", "sk": "profile"},
|
||||
UpdateExpression="SET #n = :name, age = :age",
|
||||
ExpressionAttributeNames={"#n": "name"}, # "name" is a reserved word
|
||||
ExpressionAttributeValues={":name": "Bob", ":age": 31},
|
||||
)
|
||||
|
||||
# Delete item
|
||||
table.delete_item(Key={"pk": "user#1", "sk": "profile"})
|
||||
|
||||
# Conditional write
|
||||
table.put_item(
|
||||
Item={"pk": "user#1", "sk": "profile", "name": "Alice"},
|
||||
ConditionExpression=Attr("pk").not_exists(), # only if item doesn't exist
|
||||
)
|
||||
```
|
||||
|
||||
### Query
|
||||
|
||||
```python
|
||||
# Query by partition key
|
||||
response = table.query(
|
||||
KeyConditionExpression=Key("pk").eq("user#1"),
|
||||
)
|
||||
|
||||
# Query with sort key condition
|
||||
response = table.query(
|
||||
KeyConditionExpression=Key("pk").eq("user#1") & Key("sk").begins_with("order#"),
|
||||
)
|
||||
|
||||
# Query with filter (applied after read, still consumes RCUs)
|
||||
response = table.query(
|
||||
KeyConditionExpression=Key("pk").eq("user#1"),
|
||||
FilterExpression=Attr("status").eq("active"),
|
||||
)
|
||||
|
||||
# Query a GSI
|
||||
response = table.query(
|
||||
IndexName="gsi-email",
|
||||
KeyConditionExpression=Key("email").eq("alice@example.com"),
|
||||
)
|
||||
|
||||
# Reverse order
|
||||
response = table.query(
|
||||
KeyConditionExpression=Key("pk").eq("user#1"),
|
||||
ScanIndexForward=False, # descending sort key order
|
||||
)
|
||||
|
||||
# Projection -- return only specific attributes
|
||||
response = table.query(
|
||||
KeyConditionExpression=Key("pk").eq("user#1"),
|
||||
ProjectionExpression="pk, sk, #n",
|
||||
ExpressionAttributeNames={"#n": "name"},
|
||||
)
|
||||
```
|
||||
|
||||
### Scan
|
||||
|
||||
```python
|
||||
# Full table scan (expensive -- prefer query when possible)
|
||||
response = table.scan()
|
||||
items = response["Items"]
|
||||
|
||||
# Scan with filter
|
||||
response = table.scan(
|
||||
FilterExpression=Attr("age").gte(18) & Attr("status").eq("active"),
|
||||
)
|
||||
```
|
||||
|
||||
### Batch Operations
|
||||
|
||||
```python
|
||||
# Batch write -- auto-chunks into 25-item batches, retries unprocessed items
|
||||
with table.batch_writer() as batch:
|
||||
for item in items:
|
||||
batch.put_item(Item=item)
|
||||
|
||||
# Can also delete
|
||||
batch.delete_item(Key={"pk": "user#old", "sk": "profile"})
|
||||
|
||||
# Batch get (across tables) -- use the resource, not table
|
||||
dynamodb = boto3.resource("dynamodb")
|
||||
response = dynamodb.batch_get_item(
|
||||
RequestItems={
|
||||
"my-table": {
|
||||
"Keys": [
|
||||
{"pk": "user#1", "sk": "profile"},
|
||||
{"pk": "user#2", "sk": "profile"},
|
||||
],
|
||||
}
|
||||
}
|
||||
)
|
||||
items = response["Responses"]["my-table"]
|
||||
```
|
||||
|
||||
## Condition Expressions
|
||||
|
||||
Always use `Key` and `Attr` condition builders with the resource interface. Never hand-build expression strings or manually construct `ExpressionAttributeNames`/`ExpressionAttributeValues` when a condition builder can do it:
|
||||
|
||||
```python
|
||||
# Right -- condition builders handle serialization and placeholders
|
||||
table.put_item(
|
||||
Item=item,
|
||||
ConditionExpression=Attr("pk").not_exists(),
|
||||
)
|
||||
|
||||
# Wrong -- manual string building defeats the purpose of the resource interface
|
||||
table.put_item(
|
||||
Item=item,
|
||||
ConditionExpression="attribute_not_exists(#pk)",
|
||||
ExpressionAttributeNames={"#pk": "pk"},
|
||||
)
|
||||
```
|
||||
|
||||
```python
|
||||
from boto3.dynamodb.conditions import Key, Attr
|
||||
|
||||
# Key conditions (for KeyConditionExpression in query)
|
||||
Key("pk").eq("value")
|
||||
Key("sk").begins_with("prefix")
|
||||
Key("sk").between("a", "z")
|
||||
Key("sk").lt("value")
|
||||
Key("sk").lte("value")
|
||||
Key("sk").gt("value")
|
||||
Key("sk").gte("value")
|
||||
|
||||
# Attribute conditions (for FilterExpression and ConditionExpression)
|
||||
Attr("field").eq("value")
|
||||
Attr("field").ne("value")
|
||||
Attr("field").lt(10)
|
||||
Attr("field").lte(10)
|
||||
Attr("field").gt(10)
|
||||
Attr("field").gte(10)
|
||||
Attr("field").begins_with("prefix")
|
||||
Attr("field").between(1, 100)
|
||||
Attr("field").is_in(["a", "b", "c"])
|
||||
Attr("field").exists()
|
||||
Attr("field").not_exists()
|
||||
Attr("field").contains("substring") # works on strings, lists, and sets
|
||||
Attr("field").size()
|
||||
|
||||
# Combine with & (AND), | (OR), ~ (NOT)
|
||||
(Attr("age").gte(18)) & (Attr("status").eq("active"))
|
||||
(Attr("role").eq("admin")) | (Attr("role").eq("superadmin"))
|
||||
~Attr("deleted").exists()
|
||||
|
||||
# Nested attributes
|
||||
Attr("address.city").eq("Seattle")
|
||||
```
|
||||
|
||||
## Type Handling
|
||||
|
||||
### Resource auto-marshalling
|
||||
|
||||
The resource interface handles type conversion automatically:
|
||||
|
||||
| Python type | DynamoDB type |
|
||||
|---|---|
|
||||
| `str` | S |
|
||||
| `int`, `Decimal` | N |
|
||||
| `bytes`, `bytearray` | B |
|
||||
| `bool` | BOOL |
|
||||
| `None` | NULL |
|
||||
| `list` | L |
|
||||
| `dict` | M |
|
||||
| `set` of `str` | SS |
|
||||
| `set` of `int`/`Decimal` | NS |
|
||||
| `set` of `bytes` | BS |
|
||||
|
||||
Use `Decimal` for numbers when precision matters. DynamoDB stores numbers as strings internally, and `float` values may introduce floating-point precision artifacts:
|
||||
|
||||
```python
|
||||
from decimal import Decimal
|
||||
|
||||
# Exact representation
|
||||
table.put_item(Item={"pk": "1", "price": Decimal("19.99")})
|
||||
|
||||
# Works but may lose precision -- float 19.99 is stored as
|
||||
# Decimal("19.9900000000000002131628...") internally
|
||||
table.put_item(Item={"pk": "1", "price": 19.99})
|
||||
```
|
||||
|
||||
### Client interface (manual marshalling)
|
||||
|
||||
If you must use the client interface, use `TypeSerializer`/`TypeDeserializer`:
|
||||
|
||||
```python
|
||||
from boto3.dynamodb.types import TypeSerializer, TypeDeserializer
|
||||
|
||||
serializer = TypeSerializer()
|
||||
deserializer = TypeDeserializer()
|
||||
|
||||
# Serialize a Python value to DynamoDB format
|
||||
serializer.serialize("hello") # {"S": "hello"}
|
||||
serializer.serialize(42) # {"N": "42"}
|
||||
serializer.serialize(True) # {"BOOL": True}
|
||||
|
||||
# Deserialize DynamoDB format to Python value
|
||||
deserializer.deserialize({"S": "hello"}) # "hello"
|
||||
deserializer.deserialize({"N": "42"}) # Decimal("42")
|
||||
```
|
||||
|
||||
## Pagination (Query / Scan)
|
||||
|
||||
DynamoDB returns up to 1MB per call. Use the resource's underlying client to get paginators with auto-marshalled types:
|
||||
|
||||
```python
|
||||
dynamodb = boto3.resource("dynamodb").meta.client
|
||||
paginator = dynamodb.get_paginator("query")
|
||||
for page in paginator.paginate(
|
||||
TableName="my-table",
|
||||
KeyConditionExpression="pk = :pk",
|
||||
ExpressionAttributeValues={":pk": "user#1"}, # auto-marshalled, no {"S": ...}
|
||||
):
|
||||
for item in page["Items"]:
|
||||
print(item)
|
||||
```
|
||||
@@ -0,0 +1,139 @@
|
||||
# Error Handling Reference
|
||||
|
||||
## Core Principle
|
||||
|
||||
Only catch an exception when you have an actionable response: return a fallback, retry, take a different code path. If the only thing you'd do is print the error, don't catch it -- let it propagate. The caller (or a top-level handler) is in a better position to decide what to do.
|
||||
|
||||
## ClientError Anatomy
|
||||
|
||||
`botocore.exceptions.ClientError` is the base exception for all AWS API errors:
|
||||
|
||||
```python
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
try:
|
||||
client.describe_instances(InstanceIds=["i-nonexistent"])
|
||||
except ClientError as e:
|
||||
error = e.response["Error"]
|
||||
metadata = e.response["ResponseMetadata"]
|
||||
|
||||
error["Code"] # "InvalidInstanceID.NotFound"
|
||||
error["Message"] # "The instance ID 'i-nonexistent' does not exist"
|
||||
metadata["HTTPStatusCode"] # 400
|
||||
metadata["RequestId"] # AWS request ID for support cases
|
||||
```
|
||||
|
||||
## Service-Specific Exceptions
|
||||
|
||||
Each client exposes typed exceptions generated from the service model. These are subclasses of `ClientError`, so a `ClientError` catch still works as a fallback:
|
||||
|
||||
```python
|
||||
s3 = boto3.client("s3")
|
||||
try:
|
||||
s3.get_object(Bucket="bucket", Key="key")
|
||||
except s3.exceptions.NoSuchKey:
|
||||
return None # actionable: missing key is a valid case
|
||||
```
|
||||
|
||||
List available exceptions for a client:
|
||||
|
||||
```python
|
||||
print([e for e in dir(s3.exceptions) if not e.startswith("_")])
|
||||
```
|
||||
|
||||
## Common botocore Exceptions
|
||||
|
||||
```python
|
||||
from botocore.exceptions import (
|
||||
ClientError, # AWS API returned an error response
|
||||
NoCredentialsError, # no credentials found in the chain
|
||||
PartialCredentialsError, # incomplete credentials (e.g. key without secret)
|
||||
NoRegionError, # no region configured
|
||||
ParamValidationError, # invalid parameters before request is sent
|
||||
EndpointConnectionError, # could not connect to the endpoint
|
||||
ConnectTimeoutError, # connection timed out
|
||||
ReadTimeoutError, # read timed out waiting for response
|
||||
WaiterError, # waiter reached max attempts without success
|
||||
)
|
||||
```
|
||||
|
||||
`ParamValidationError` is raised locally before any network request -- it means the parameters failed botocore's client-side validation.
|
||||
|
||||
## Error Handling Patterns
|
||||
|
||||
### Actionable catch: convert to return value
|
||||
|
||||
```python
|
||||
def get_item(table, key: dict) -> dict | None:
|
||||
response = table.get_item(Key=key)
|
||||
return response.get("Item") # None if missing, no exception needed
|
||||
|
||||
def head_object(client, bucket: str, key: str) -> dict | None:
|
||||
try:
|
||||
return client.head_object(Bucket=bucket, Key=key)
|
||||
except client.exceptions.ClientError as e:
|
||||
if e.response["ResponseMetadata"]["HTTPStatusCode"] == 404:
|
||||
return None
|
||||
raise
|
||||
```
|
||||
|
||||
### Actionable catch: conditional put race
|
||||
|
||||
```python
|
||||
try:
|
||||
table.put_item(
|
||||
Item=new_item,
|
||||
ConditionExpression=Attr("pk").not_exists(),
|
||||
)
|
||||
except table.meta.client.exceptions.ConditionalCheckFailedException:
|
||||
# Another writer got there first -- fetch what they wrote
|
||||
return table.get_item(Key={"pk": new_item["pk"]})["Item"]
|
||||
```
|
||||
|
||||
### Actionable catch: create-if-not-exists
|
||||
|
||||
```python
|
||||
try:
|
||||
client.create_bucket(Bucket="my-bucket")
|
||||
except client.exceptions.BucketAlreadyOwnedByYou:
|
||||
pass # already exists, that's fine
|
||||
```
|
||||
|
||||
### Top-level catch-all in main()
|
||||
|
||||
Business logic functions should let exceptions propagate. The `main()` function is the right place for a generic catch-all that presents errors cleanly to the user. Keep the catch-all simple -- just `ClientError`. Other exceptions like `NoCredentialsError` already have clear messages and can propagate naturally:
|
||||
|
||||
```python
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
do_the_work()
|
||||
return 0
|
||||
except ClientError as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
```
|
||||
|
||||
### What NOT to do
|
||||
|
||||
```python
|
||||
# Wrong: catching just to print and swallow
|
||||
try:
|
||||
client.describe_table(TableName=name)
|
||||
except client.exceptions.ResourceNotFoundException:
|
||||
print("Table not found") # swallowed -- caller has no idea it failed
|
||||
except NoCredentialsError:
|
||||
print("No credentials") # swallowed
|
||||
except EndpointConnectionError:
|
||||
print("Can't connect") # swallowed
|
||||
|
||||
# Wrong: sys.exit() from a business logic function
|
||||
def process_queue(queue_url):
|
||||
if not queue_url:
|
||||
print("No queue URL provided")
|
||||
sys.exit(1) # untestable, unusable as library code
|
||||
```
|
||||
@@ -0,0 +1,104 @@
|
||||
# Pagination Reference
|
||||
|
||||
## Paginators
|
||||
|
||||
Most `list_*`, `describe_*`, and `get_*` operations that return collections support pagination. When you only need specific fields, use `.search()` to extract and flatten across pages:
|
||||
|
||||
```python
|
||||
client = boto3.client("ec2")
|
||||
paginator = client.get_paginator("describe_instances")
|
||||
|
||||
for instance_id in paginator.paginate().search("Reservations[].Instances[].InstanceId"):
|
||||
print(instance_id)
|
||||
```
|
||||
|
||||
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 reservation in page.get("Reservations", []):
|
||||
for instance in reservation.get("Instances", []):
|
||||
process(instance)
|
||||
```
|
||||
|
||||
Check if an operation supports pagination:
|
||||
|
||||
```python
|
||||
client.can_paginate("describe_instances") # True
|
||||
```
|
||||
|
||||
## Pagination Configuration
|
||||
|
||||
Control page size and total items via `PaginationConfig`:
|
||||
|
||||
```python
|
||||
paginator = client.get_paginator("list_objects_v2")
|
||||
pages = paginator.paginate(
|
||||
Bucket="my-bucket",
|
||||
PaginationConfig={
|
||||
"PageSize": 100, # items per API call
|
||||
"MaxItems": 500, # total items across all pages
|
||||
"StartingToken": None, # resume from a previous NextToken
|
||||
},
|
||||
)
|
||||
```
|
||||
|
||||
- `PageSize` controls the `MaxKeys`/`MaxResults`/`Limit` parameter sent to the API
|
||||
- `MaxItems` stops iteration after this many total items and provides a `NextToken` for resuming
|
||||
- The paginator uses the correct token parameter name for each service automatically
|
||||
|
||||
## JMESPath Filtering
|
||||
|
||||
Use `.search()` to extract and flatten results across pages:
|
||||
|
||||
```python
|
||||
paginator = client.get_paginator("list_objects_v2")
|
||||
page_iterator = paginator.paginate(Bucket="my-bucket")
|
||||
|
||||
# Flatten all keys across all pages
|
||||
keys = page_iterator.search("Contents[].Key")
|
||||
for key in keys:
|
||||
print(key)
|
||||
|
||||
# Filter with JMESPath expressions
|
||||
large_objects = page_iterator.search(
|
||||
"Contents[?Size > `1048576`].{Key: Key, Size: Size}"
|
||||
)
|
||||
```
|
||||
|
||||
`.search()` returns a generator that yields individual items, not pages -- no need to handle page boundaries.
|
||||
|
||||
## Common Paginated Operations
|
||||
|
||||
| Service | Operation | Result key |
|
||||
|---|---|---|
|
||||
| S3 | `list_objects_v2` | `Contents` |
|
||||
| DynamoDB | `scan` | `Items` |
|
||||
| DynamoDB | `query` | `Items` |
|
||||
| EC2 | `describe_instances` | `Reservations` |
|
||||
| IAM | `list_users` | `Users` |
|
||||
| Lambda | `list_functions` | `Functions` |
|
||||
| CloudWatch Logs | `describe_log_groups` | `logGroups` |
|
||||
|
||||
Note: `list_buckets` is not paginated -- it returns all buckets in a single response.
|
||||
|
||||
## Resource-Level Pagination
|
||||
|
||||
Resources handle pagination automatically via collection methods:
|
||||
|
||||
```python
|
||||
s3 = boto3.resource("s3")
|
||||
bucket = s3.Bucket("my-bucket")
|
||||
|
||||
# .all() paginates automatically
|
||||
for obj in bucket.objects.all():
|
||||
print(obj.key)
|
||||
|
||||
# .filter() also paginates
|
||||
for obj in bucket.objects.filter(Prefix="logs/"):
|
||||
print(obj.key)
|
||||
|
||||
# .limit() limits total results
|
||||
for obj in bucket.objects.limit(100):
|
||||
print(obj.key)
|
||||
```
|
||||
@@ -0,0 +1,223 @@
|
||||
# S3 Reference
|
||||
|
||||
## Common Operations
|
||||
|
||||
Use transfer methods for file upload/download -- they handle multipart automatically:
|
||||
|
||||
```python
|
||||
import boto3
|
||||
|
||||
s3 = boto3.client("s3")
|
||||
|
||||
# Upload / download files
|
||||
s3.upload_file("local.txt", "my-bucket", "remote.txt")
|
||||
s3.download_file("my-bucket", "remote.txt", "local.txt")
|
||||
|
||||
# Upload / download file-like objects
|
||||
with open("local.txt", "rb") as f:
|
||||
s3.upload_fileobj(f, "my-bucket", "remote.txt")
|
||||
|
||||
# Presigned URL
|
||||
url = s3.generate_presigned_url(
|
||||
"get_object",
|
||||
Params={"Bucket": "my-bucket", "Key": "my-key"},
|
||||
ExpiresIn=3600,
|
||||
)
|
||||
```
|
||||
|
||||
**Always close S3 streaming bodies** -- unread `Body` streams hold connections open:
|
||||
|
||||
```python
|
||||
response = s3.get_object(Bucket="bucket", Key="key")
|
||||
try:
|
||||
data = response["Body"].read()
|
||||
finally:
|
||||
response["Body"].close()
|
||||
|
||||
# Or use as context manager
|
||||
with s3.get_object(Bucket="bucket", Key="key")["Body"] as body:
|
||||
data = body.read()
|
||||
```
|
||||
|
||||
## Transfer Methods
|
||||
|
||||
The S3 client and resource provide managed transfer methods that handle
|
||||
multipart upload/download, retries, and parallelism automatically:
|
||||
|
||||
```python
|
||||
import boto3
|
||||
|
||||
s3 = boto3.client("s3")
|
||||
|
||||
# File transfers
|
||||
s3.upload_file("local.txt", "my-bucket", "remote.txt")
|
||||
s3.download_file("my-bucket", "remote.txt", "local.txt")
|
||||
|
||||
# File-like object transfers
|
||||
with open("local.txt", "rb") as f:
|
||||
s3.upload_fileobj(f, "my-bucket", "remote.txt")
|
||||
|
||||
with open("local.txt", "wb") as f:
|
||||
s3.download_fileobj("my-bucket", "remote.txt", f)
|
||||
```
|
||||
|
||||
### Extra arguments
|
||||
|
||||
Pass any PutObject/GetObject parameters via `ExtraArgs`:
|
||||
|
||||
```python
|
||||
s3.upload_file(
|
||||
"local.txt", "my-bucket", "remote.txt",
|
||||
ExtraArgs={
|
||||
"ContentType": "text/plain",
|
||||
"ServerSideEncryption": "aws:kms",
|
||||
"Metadata": {"author": "alice"},
|
||||
},
|
||||
)
|
||||
```
|
||||
|
||||
### Progress callbacks
|
||||
|
||||
```python
|
||||
import os
|
||||
|
||||
file_size = os.path.getsize("large_file.bin")
|
||||
uploaded = 0
|
||||
|
||||
def progress(bytes_transferred):
|
||||
nonlocal uploaded
|
||||
uploaded += bytes_transferred
|
||||
pct = (uploaded / file_size) * 100
|
||||
print(f"\r{pct:.1f}%", end="")
|
||||
|
||||
s3.upload_file("large_file.bin", "bucket", "key", Callback=progress)
|
||||
```
|
||||
|
||||
## TransferConfig
|
||||
|
||||
Control multipart thresholds and concurrency:
|
||||
|
||||
```python
|
||||
from boto3.s3.transfer import TransferConfig
|
||||
|
||||
config = TransferConfig(
|
||||
multipart_threshold=8 * 1024 * 1024, # switch to multipart above 8MB (default 8MB)
|
||||
max_concurrency=10, # parallel transfer threads (default 10)
|
||||
multipart_chunksize=8 * 1024 * 1024, # part size (default 8MB, min 5MB)
|
||||
use_threads=True, # enable threading (default True)
|
||||
)
|
||||
|
||||
s3.upload_file("large.bin", "bucket", "key", Config=config)
|
||||
s3.download_file("bucket", "key", "large.bin", Config=config)
|
||||
```
|
||||
|
||||
## Streaming Body
|
||||
|
||||
`get_object` returns a `StreamingBody` that must be read or closed:
|
||||
|
||||
```python
|
||||
response = s3.get_object(Bucket="bucket", Key="key")
|
||||
body = response["Body"]
|
||||
|
||||
# Read all at once
|
||||
data = body.read()
|
||||
body.close()
|
||||
|
||||
# Read in chunks
|
||||
for chunk in body.iter_chunks(chunk_size=4096):
|
||||
process(chunk)
|
||||
body.close()
|
||||
|
||||
# Read lines (for text content)
|
||||
for line in body.iter_lines():
|
||||
process(line)
|
||||
|
||||
# As context manager -- auto-closes
|
||||
with s3.get_object(Bucket="bucket", Key="key")["Body"] as body:
|
||||
data = body.read()
|
||||
```
|
||||
|
||||
`StreamingBody` can only be read once. If you need the data multiple times, save it to a variable.
|
||||
|
||||
For file downloads, prefer `download_file`/`download_fileobj` over `get_object` -- they handle multipart, retries, and stream cleanup automatically.
|
||||
|
||||
## Presigned URLs
|
||||
|
||||
### GET (download)
|
||||
|
||||
```python
|
||||
url = s3.generate_presigned_url(
|
||||
"get_object",
|
||||
Params={"Bucket": "bucket", "Key": "key"},
|
||||
ExpiresIn=3600, # seconds (default 3600)
|
||||
)
|
||||
```
|
||||
|
||||
### PUT (upload)
|
||||
|
||||
```python
|
||||
url = s3.generate_presigned_url(
|
||||
"put_object",
|
||||
Params={
|
||||
"Bucket": "bucket",
|
||||
"Key": "key",
|
||||
"ContentType": "application/pdf",
|
||||
},
|
||||
ExpiresIn=3600,
|
||||
)
|
||||
# Client must include Content-Type: application/pdf in the upload request
|
||||
```
|
||||
|
||||
### POST (browser form upload)
|
||||
|
||||
```python
|
||||
presigned = s3.generate_presigned_post(
|
||||
Bucket="bucket",
|
||||
Key="uploads/${filename}",
|
||||
Conditions=[
|
||||
["content-length-range", 0, 10 * 1024 * 1024], # max 10MB
|
||||
{"Content-Type": "image/jpeg"},
|
||||
],
|
||||
Fields={"Content-Type": "image/jpeg"},
|
||||
ExpiresIn=600,
|
||||
)
|
||||
# presigned["url"] -- POST URL
|
||||
# presigned["fields"] -- form fields to include
|
||||
```
|
||||
|
||||
## Copy Operations
|
||||
|
||||
```python
|
||||
# Client -- simple copy
|
||||
s3.copy_object(
|
||||
Bucket="dest-bucket",
|
||||
Key="dest-key",
|
||||
CopySource={"Bucket": "src-bucket", "Key": "src-key"},
|
||||
)
|
||||
|
||||
# Resource -- handles multipart for large objects automatically
|
||||
s3_resource = boto3.resource("s3")
|
||||
copy_source = {"Bucket": "src-bucket", "Key": "src-key"}
|
||||
s3_resource.Object("dest-bucket", "dest-key").copy(copy_source)
|
||||
```
|
||||
|
||||
## Resource Interface
|
||||
|
||||
```python
|
||||
s3 = boto3.resource("s3")
|
||||
|
||||
# Bucket operations
|
||||
bucket = s3.Bucket("my-bucket")
|
||||
for obj in bucket.objects.filter(Prefix="logs/"):
|
||||
print(obj.key, obj.size)
|
||||
|
||||
# Object operations
|
||||
obj = s3.Object("my-bucket", "my-key")
|
||||
obj.upload_file("local.txt")
|
||||
obj.download_file("local.txt")
|
||||
obj.delete()
|
||||
|
||||
# Read object body
|
||||
response = obj.get()
|
||||
data = response["Body"].read()
|
||||
```
|
||||
@@ -0,0 +1,121 @@
|
||||
# Waiters Reference
|
||||
|
||||
## Using Waiters
|
||||
|
||||
Waiters poll an AWS operation until a resource reaches a desired state or the waiter times out:
|
||||
|
||||
```python
|
||||
import boto3
|
||||
|
||||
ec2 = boto3.client("ec2")
|
||||
|
||||
# Start an instance
|
||||
ec2.start_instances(InstanceIds=["i-1234567890abcdef0"])
|
||||
|
||||
# Wait until it's running
|
||||
waiter = ec2.get_waiter("instance_running")
|
||||
waiter.wait(
|
||||
InstanceIds=["i-1234567890abcdef0"],
|
||||
WaiterConfig={
|
||||
"Delay": 15, # seconds between polls (default varies by waiter)
|
||||
"MaxAttempts": 40, # max poll attempts (default varies by waiter)
|
||||
},
|
||||
)
|
||||
```
|
||||
|
||||
## WaiterConfig
|
||||
|
||||
| Parameter | Description |
|
||||
|---|---|
|
||||
| `Delay` | Seconds between polling attempts |
|
||||
| `MaxAttempts` | Maximum number of polling attempts before raising `WaiterError` |
|
||||
|
||||
Both are optional and override the waiter's built-in defaults.
|
||||
|
||||
## Common Waiters
|
||||
|
||||
| Service | Waiter | Polls until |
|
||||
|---|---|---|
|
||||
| S3 | `bucket_exists` | HeadBucket succeeds |
|
||||
| S3 | `bucket_not_exists` | HeadBucket returns 404 |
|
||||
| S3 | `object_exists` | HeadObject succeeds |
|
||||
| S3 | `object_not_exists` | HeadObject returns 404 |
|
||||
| EC2 | `instance_running` | Instance state is "running" |
|
||||
| EC2 | `instance_stopped` | Instance state is "stopped" |
|
||||
| EC2 | `instance_terminated` | Instance state is "terminated" |
|
||||
| RDS | `db_instance_available` | DB instance is "available" |
|
||||
| CloudFormation | `stack_create_complete` | Stack status is CREATE_COMPLETE |
|
||||
| CloudFormation | `stack_delete_complete` | Stack no longer exists |
|
||||
|
||||
List available waiters for a client:
|
||||
|
||||
```python
|
||||
client.waiter_names # ["bucket_exists", "bucket_not_exists", ...]
|
||||
```
|
||||
|
||||
## Waiter Errors
|
||||
|
||||
```python
|
||||
from botocore.exceptions import WaiterError
|
||||
|
||||
try:
|
||||
waiter = s3.get_waiter("object_exists")
|
||||
waiter.wait(Bucket="bucket", Key="key")
|
||||
except WaiterError as e:
|
||||
print(f"Waiter failed: {e}")
|
||||
# e.last_response contains the last polling response
|
||||
```
|
||||
|
||||
A `WaiterError` is raised when:
|
||||
|
||||
- `MaxAttempts` is exceeded without reaching the desired state
|
||||
- The waiter enters a terminal failure state (e.g., the resource entered an unrecoverable state)
|
||||
|
||||
## Custom Waiters
|
||||
|
||||
For operations without built-in waiters, define a custom waiter model:
|
||||
|
||||
```python
|
||||
import boto3
|
||||
from botocore.waiter import WaiterModel, create_waiter_with_client
|
||||
|
||||
waiter_config = {
|
||||
"version": 2,
|
||||
"waiters": {
|
||||
"FunctionActive": {
|
||||
"operation": "GetFunction",
|
||||
"delay": 5,
|
||||
"maxAttempts": 20,
|
||||
"acceptors": [
|
||||
{
|
||||
"matcher": "path",
|
||||
"expected": "Active",
|
||||
"argument": "Configuration.State",
|
||||
"state": "success",
|
||||
},
|
||||
{
|
||||
"matcher": "path",
|
||||
"expected": "Failed",
|
||||
"argument": "Configuration.State",
|
||||
"state": "failure",
|
||||
},
|
||||
],
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
client = boto3.client("lambda")
|
||||
waiter_model = WaiterModel(waiter_config)
|
||||
waiter = create_waiter_with_client("FunctionActive", waiter_model, client)
|
||||
waiter.wait(FunctionName="my-function")
|
||||
```
|
||||
|
||||
### Acceptor matchers
|
||||
|
||||
| Matcher | Description |
|
||||
|---|---|
|
||||
| `path` | JMESPath expression against the response |
|
||||
| `pathAll` | All items in a JMESPath list must match |
|
||||
| `pathAny` | Any item in a JMESPath list must match |
|
||||
| `status` | HTTP status code |
|
||||
| `error` | Error code string |
|
||||
Reference in New Issue
Block a user