Aws Sdk Python Usage

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).

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

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

# 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:

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:

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:

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:

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:

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:

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.:

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:

# 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

Client Configuration Reference

botocore.config.Config

All configuration is passed via botocore.config.Config. Multiple configs can be merged:

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

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).

Can also set via AWS_MAX_ATTEMPTS and AWS_RETRY_MODE env vars or ~/.aws/config.

Timeouts

config = Config(
    connect_timeout=5,    # seconds to establish connection (default 60)
    read_timeout=10,      # seconds to wait for response data (default 60)
)

Connection Pool

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

# 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

# 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

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

config = Config(
    user_agent_appid="my-app/1.0",
    user_agent_extra="custom-metadata",
)

Sharing Config Across Clients

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:

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')

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

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

# 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)

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:

[profile cross-account]
role_arn = arn:aws:iam::123456789012:role/MyRole
source_profile = default

session = boto3.Session(profile_name="cross-account")
client = session.client("s3")  # credentials auto-refresh

Chained Role Assumption

# ~/.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

STS Get Caller Identity

Useful for verifying which credentials are in use:

sts = boto3.client("sts")
identity = sts.get_caller_identity()
print(identity["Account"], identity["Arn"])

DynamoDB Reference

Use the resource interface to work with native Python types instead of AttributeValue dicts:

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:

# 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):

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:

import boto3
from boto3.dynamodb.conditions import Key, Attr
from decimal import Decimal

table = boto3.resource("dynamodb").Table("my-table")

CRUD Operations

# 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

# 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

# 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

# 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:

# 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"},
)

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:

Use Decimal for numbers when precision matters. DynamoDB stores numbers as strings internally, and float values may introduce floating-point precision artifacts:

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:

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:

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)

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:

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:

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:

print([e for e in dir(s3.exceptions) if not e.startswith("_")])

Common botocore Exceptions

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

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

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

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:

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

# 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

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:

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:

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:

client.can_paginate("describe_instances")  # True

Pagination Configuration

Control page size and total items via PaginationConfig:

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:

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

Note: list_buckets is not paginated -- it returns all buckets in a single response.

Resource-Level Pagination

Resources handle pagination automatically via collection methods:

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)

S3 Reference

Common Operations

Use transfer methods for file upload/download -- they handle multipart automatically:

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:

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:

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:

s3.upload_file(
    "local.txt", "my-bucket", "remote.txt",
    ExtraArgs={
        "ContentType": "text/plain",
        "ServerSideEncryption": "aws:kms",
        "Metadata": {"author": "alice"},
    },
)

Progress callbacks

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:

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:

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)

url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "bucket", "Key": "key"},
    ExpiresIn=3600,  # seconds (default 3600)
)

PUT (upload)

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)

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

# 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

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()

Waiters Reference

Using Waiters

Waiters poll an AWS operation until a resource reaches a desired state or the waiter times out:

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

Both are optional and override the waiter's built-in defaults.

Common Waiters

List available waiters for a client:

client.waiter_names  # ["bucket_exists", "bucket_not_exists", ...]

Waiter Errors

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:

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