Boto3

Python boto3 cheatsheet for common AWS services and daily automation tasks.

Boto3 Cheatsheet

Install & Setup

Install

pip install boto3 botocore

Basic Session

import boto3
from botocore.exceptions import ClientError

session = boto3.Session(
    profile_name="dev",
    region_name="us-east-1",
)

s3 = session.client("s3")
ec2 = session.resource("ec2")

Clients vs Resources

# Client: low-level API, available for every AWS service
s3_client = boto3.client("s3")
s3_client.list_buckets()

# Resource: higher-level object API, available for selected services
s3_resource = boto3.resource("s3")
bucket = s3_resource.Bucket("my-bucket")

Credentials Resolution Order

Common places boto3 checks:

Explicit parameters in code.
Environment variables such as AWS_ACCESS_KEY_ID.
Shared credentials file: ~/.aws/credentials.
Shared config file: ~/.aws/config.
Assume-role profiles.
Container or EC2 instance role metadata.

Common Imports

import json
import time
import boto3

from pathlib import Path
from botocore.config import Config
from botocore.exceptions import ClientError, NoCredentialsError, WaiterError

Retry & Timeout Config

from botocore.config import Config

config = Config(
    retries={"max_attempts": 10, "mode": "standard"},
    connect_timeout=5,
    read_timeout=60,
)

s3 = boto3.client("s3", config=config)

Error Handling

Catch AWS API Errors

from botocore.exceptions import ClientError

try:
    s3.head_bucket(Bucket="my-bucket")
except ClientError as exc:
    code = exc.response["Error"]["Code"]
    message = exc.response["Error"]["Message"]
    print(code, message)

Match Specific Errors

try:
    dynamodb.get_item(
        TableName="users",
        Key={"pk": {"S": "user#1"}},
    )
except ClientError as exc:
    if exc.response["Error"]["Code"] == "ResourceNotFoundException":
        print("Table does not exist")
    else:
        raise

Pagination & Waiters

Paginators

s3 = boto3.client("s3")
paginator = s3.get_paginator("list_objects_v2")

for page in paginator.paginate(Bucket="my-bucket", Prefix="logs/"):
    for obj in page.get("Contents", []):
        print(obj["Key"], obj["Size"])

Waiters

ec2 = boto3.client("ec2")

ec2.start_instances(InstanceIds=["i-0123456789abcdef0"])

waiter = ec2.get_waiter("instance_running")
waiter.wait(InstanceIds=["i-0123456789abcdef0"])

Amazon S3

List Buckets

s3 = boto3.client("s3")

for bucket in s3.list_buckets()["Buckets"]:
    print(bucket["Name"], bucket["CreationDate"])

Create Bucket

s3.create_bucket(
    Bucket="my-unique-bucket-name",
    CreateBucketConfiguration={"LocationConstraint": "ap-south-1"},
)

For us-east-1, omit CreateBucketConfiguration.

Upload File

s3.upload_file(
    Filename="report.csv",
    Bucket="my-bucket",
    Key="reports/report.csv",
)

Upload Bytes or JSON

payload = {"status": "ok"}

s3.put_object(
    Bucket="my-bucket",
    Key="data/status.json",
    Body=json.dumps(payload).encode("utf-8"),
    ContentType="application/json",
)

Download File

s3.download_file(
    Bucket="my-bucket",
    Key="reports/report.csv",
    Filename="report.csv",
)

Read Object

response = s3.get_object(Bucket="my-bucket", Key="data/status.json")
data = json.loads(response["Body"].read())

List Objects

paginator = s3.get_paginator("list_objects_v2")

for page in paginator.paginate(Bucket="my-bucket", Prefix="logs/"):
    for obj in page.get("Contents", []):
        print(obj["Key"])

Copy Object

s3.copy_object(
    Bucket="target-bucket",
    Key="archive/report.csv",
    CopySource={"Bucket": "source-bucket", "Key": "reports/report.csv"},
)

Delete Object

s3.delete_object(Bucket="my-bucket", Key="reports/old.csv")

Delete Many Objects

objects = [{"Key": "tmp/a.txt"}, {"Key": "tmp/b.txt"}]

s3.delete_objects(
    Bucket="my-bucket",
    Delete={"Objects": objects, "Quiet": True},
)

Generate Presigned URL

url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "my-bucket", "Key": "reports/report.csv"},
    ExpiresIn=3600,
)
print(url)

Amazon EC2

List Instances

ec2 = boto3.client("ec2")

response = ec2.describe_instances()

for reservation in response["Reservations"]:
    for instance in reservation["Instances"]:
        print(instance["InstanceId"], instance["State"]["Name"])

Filter Instances

response = ec2.describe_instances(
    Filters=[
        {"Name": "instance-state-name", "Values": ["running"]},
        {"Name": "tag:Environment", "Values": ["prod"]},
    ]
)

Start, Stop, Reboot

ids = ["i-0123456789abcdef0"]

ec2.start_instances(InstanceIds=ids)
ec2.stop_instances(InstanceIds=ids)
ec2.reboot_instances(InstanceIds=ids)

Launch Instance

response = ec2.run_instances(
    ImageId="ami-0123456789abcdef0",
    InstanceType="t3.micro",
    MinCount=1,
    MaxCount=1,
    KeyName="my-key",
    SecurityGroupIds=["sg-0123456789abcdef0"],
    SubnetId="subnet-0123456789abcdef0",
    TagSpecifications=[
        {
            "ResourceType": "instance",
            "Tags": [{"Key": "Name", "Value": "demo-server"}],
        }
    ],
)

instance_id = response["Instances"][0]["InstanceId"]

Terminate Instance

ec2.terminate_instances(InstanceIds=["i-0123456789abcdef0"])

Get Instance Public IPs

for reservation in ec2.describe_instances()["Reservations"]:
    for instance in reservation["Instances"]:
        print(instance["InstanceId"], instance.get("PublicIpAddress"))

Security Groups

ec2.authorize_security_group_ingress(
    GroupId="sg-0123456789abcdef0",
    IpPermissions=[
        {
            "IpProtocol": "tcp",
            "FromPort": 443,
            "ToPort": 443,
            "IpRanges": [{"CidrIp": "0.0.0.0/0", "Description": "HTTPS"}],
        }
    ],
)

IAM

List Users and Roles

iam = boto3.client("iam")

for user in iam.get_paginator("list_users").paginate():
    for item in user["Users"]:
        print(item["UserName"])

for role_page in iam.get_paginator("list_roles").paginate():
    for role in role_page["Roles"]:
        print(role["RoleName"])

Create User

iam.create_user(UserName="alice")

Attach Managed Policy

iam.attach_user_policy(
    UserName="alice",
    PolicyArn="arn:aws:iam::aws:policy/ReadOnlyAccess",
)

Create Role

assume_role_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {"Service": "lambda.amazonaws.com"},
            "Action": "sts:AssumeRole",
        }
    ],
}

iam.create_role(
    RoleName="lambda-basic-role",
    AssumeRolePolicyDocument=json.dumps(assume_role_policy),
)

Put Inline Policy

policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": ["s3:GetObject"],
            "Resource": "arn:aws:s3:::my-bucket/*",
        }
    ],
}

iam.put_role_policy(
    RoleName="lambda-basic-role",
    PolicyName="read-bucket",
    PolicyDocument=json.dumps(policy),
)

AWS STS

Caller Identity

sts = boto3.client("sts")
print(sts.get_caller_identity())

Assume Role

response = sts.assume_role(
    RoleArn="arn:aws:iam::123456789012:role/Admin",
    RoleSessionName="automation-session",
)

creds = response["Credentials"]

assumed = boto3.Session(
    aws_access_key_id=creds["AccessKeyId"],
    aws_secret_access_key=creds["SecretAccessKey"],
    aws_session_token=creds["SessionToken"],
)

AWS Lambda

List Functions

lambda_client = boto3.client("lambda")

for page in lambda_client.get_paginator("list_functions").paginate():
    for fn in page["Functions"]:
        print(fn["FunctionName"], fn["Runtime"])

Invoke Function

response = lambda_client.invoke(
    FunctionName="my-function",
    InvocationType="RequestResponse",
    Payload=json.dumps({"name": "Alice"}).encode("utf-8"),
)

payload = json.loads(response["Payload"].read())

Invoke Async

lambda_client.invoke(
    FunctionName="my-function",
    InvocationType="Event",
    Payload=json.dumps({"task": "refresh"}).encode("utf-8"),
)

Update Environment Variables

lambda_client.update_function_configuration(
    FunctionName="my-function",
    Environment={
        "Variables": {
            "ENV": "prod",
            "LOG_LEVEL": "INFO",
        }
    },
)

Update Code from S3

lambda_client.update_function_code(
    FunctionName="my-function",
    S3Bucket="deploy-artifacts",
    S3Key="lambda/my-function.zip",
    Publish=True,
)

Amazon DynamoDB

Resource Setup

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("users")

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

Update Item

response = table.update_item(
    Key={"pk": "user#1", "sk": "profile"},
    UpdateExpression="SET age = :age, updated_at = :updated_at",
    ExpressionAttributeValues={
        ":age": 31,
        ":updated_at": "2026-05-18T00:00:00Z",
    },
    ReturnValues="ALL_NEW",
)

print(response["Attributes"])

Query Partition

from boto3.dynamodb.conditions import Key

response = table.query(
    KeyConditionExpression=Key("pk").eq("user#1")
)

for item in response["Items"]:
    print(item)

Query Index

response = table.query(
    IndexName="email-index",
    KeyConditionExpression=Key("email").eq("alice@example.com"),
)

Scan with Filter

from boto3.dynamodb.conditions import Attr

response = table.scan(
    FilterExpression=Attr("status").eq("active")
)

Use scan carefully on large tables.

Batch Write

with table.batch_writer() as batch:
    for i in range(100):
        batch.put_item(Item={"pk": f"user#{i}", "sk": "profile"})

Delete Item

table.delete_item(Key={"pk": "user#1", "sk": "profile"})

Amazon SQS

Create Queue

sqs = boto3.client("sqs")

response = sqs.create_queue(
    QueueName="jobs",
    Attributes={"VisibilityTimeout": "60"},
)

queue_url = response["QueueUrl"]

Send Message

sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps({"job_id": 123}),
)

Send FIFO Message

sqs.send_message(
    QueueUrl="https://sqs.us-east-1.amazonaws.com/123/jobs.fifo",
    MessageBody="process-order",
    MessageGroupId="orders",
    MessageDeduplicationId="order-123",
)

Receive Messages

response = sqs.receive_message(
    QueueUrl=queue_url,
    MaxNumberOfMessages=10,
    WaitTimeSeconds=20,
    VisibilityTimeout=60,
)

for message in response.get("Messages", []):
    print(message["Body"])

Delete Message After Processing

sqs.delete_message(
    QueueUrl=queue_url,
    ReceiptHandle=message["ReceiptHandle"],
)

Change Visibility Timeout

sqs.change_message_visibility(
    QueueUrl=queue_url,
    ReceiptHandle=message["ReceiptHandle"],
    VisibilityTimeout=120,
)

Amazon SNS

Create Topic

sns = boto3.client("sns")

response = sns.create_topic(Name="alerts")
topic_arn = response["TopicArn"]

Publish Message

sns.publish(
    TopicArn=topic_arn,
    Subject="Build failed",
    Message="The production build failed.",
)

sns.subscribe(
    TopicArn=topic_arn,
    Protocol="email",
    Endpoint="admin@example.com",
)

Publish JSON Message

sns.publish(
    TopicArn=topic_arn,
    Message=json.dumps({"default": "Fallback", "email": "Email body"}),
    MessageStructure="json",
)

CloudWatch Logs

List Log Groups

logs = boto3.client("logs")

for page in logs.get_paginator("describe_log_groups").paginate():
    for group in page["logGroups"]:
        print(group["logGroupName"])

Read Log Events

response = logs.get_log_events(
    logGroupName="/aws/lambda/my-function",
    logStreamName="2026/05/18/[$LATEST]abcdef",
    startFromHead=True,
)

for event in response["events"]:
    print(event["message"])

Filter Logs

response = logs.filter_log_events(
    logGroupName="/aws/lambda/my-function",
    filterPattern="ERROR",
    limit=50,
)

for event in response["events"]:
    print(event["message"])

Put Metric Data

cloudwatch = boto3.client("cloudwatch")

cloudwatch.put_metric_data(
    Namespace="MyApp",
    MetricData=[
        {
            "MetricName": "JobsProcessed",
            "Value": 42,
            "Unit": "Count",
            "Dimensions": [{"Name": "Environment", "Value": "prod"}],
        }
    ],
)

AWS Systems Manager Parameter Store

Get Parameter

ssm = boto3.client("ssm")

response = ssm.get_parameter(
    Name="/myapp/prod/db-url",
    WithDecryption=True,
)

value = response["Parameter"]["Value"]

Get Parameters by Path

paginator = ssm.get_paginator("get_parameters_by_path")

for page in paginator.paginate(
    Path="/myapp/prod/",
    Recursive=True,
    WithDecryption=True,
):
    for param in page["Parameters"]:
        print(param["Name"], param["Value"])

Put Parameter

ssm.put_parameter(
    Name="/myapp/prod/api-key",
    Value="secret-value",
    Type="SecureString",
    Overwrite=True,
)

Run Command on EC2

response = ssm.send_command(
    InstanceIds=["i-0123456789abcdef0"],
    DocumentName="AWS-RunShellScript",
    Parameters={"commands": ["uptime", "df -h"]},
)

command_id = response["Command"]["CommandId"]

AWS Secrets Manager

Get Secret

secrets = boto3.client("secretsmanager")

response = secrets.get_secret_value(SecretId="prod/db")

if "SecretString" in response:
    secret = json.loads(response["SecretString"])
else:
    secret = response["SecretBinary"]

Create Secret

secrets.create_secret(
    Name="prod/api",
    SecretString=json.dumps({"token": "secret-token"}),
)

Update Secret

secrets.put_secret_value(
    SecretId="prod/api",
    SecretString=json.dumps({"token": "new-secret-token"}),
)

Amazon ECR

ecr = boto3.client("ecr")

response = ecr.get_authorization_token()
auth_data = response["authorizationData"][0]
print(auth_data["proxyEndpoint"])

For Docker login, the AWS CLI is usually simpler:

aws ecr get-login-password --region us-east-1 \
  | docker login --username AWS --password-stdin 123456789012.dkr.ecr.us-east-1.amazonaws.com

Create Repository

ecr.create_repository(
    repositoryName="my-service",
    imageScanningConfiguration={"scanOnPush": True},
)

List Images

response = ecr.list_images(repositoryName="my-service")

for image in response["imageIds"]:
    print(image)

Amazon ECS

List Clusters and Services

ecs = boto3.client("ecs")

clusters = ecs.list_clusters()["clusterArns"]

for cluster in clusters:
    services = ecs.list_services(cluster=cluster)["serviceArns"]
    print(cluster, services)

Update Service Desired Count

ecs.update_service(
    cluster="my-cluster",
    service="api",
    desiredCount=3,
)

Force New Deployment

ecs.update_service(
    cluster="my-cluster",
    service="api",
    forceNewDeployment=True,
)

Run One-Off Task

ecs.run_task(
    cluster="my-cluster",
    taskDefinition="worker:12",
    launchType="FARGATE",
    networkConfiguration={
        "awsvpcConfiguration": {
            "subnets": ["subnet-0123456789abcdef0"],
            "securityGroups": ["sg-0123456789abcdef0"],
            "assignPublicIp": "ENABLED",
        }
    },
)

Amazon RDS

List DB Instances

rds = boto3.client("rds")

for page in rds.get_paginator("describe_db_instances").paginate():
    for db in page["DBInstances"]:
        print(db["DBInstanceIdentifier"], db["DBInstanceStatus"])

Create Snapshot

rds.create_db_snapshot(
    DBInstanceIdentifier="prod-db",
    DBSnapshotIdentifier="prod-db-manual-2026-05-18",
)

Start and Stop DB Instance

rds.stop_db_instance(DBInstanceIdentifier="dev-db")
rds.start_db_instance(DBInstanceIdentifier="dev-db")

Modify DB Instance

rds.modify_db_instance(
    DBInstanceIdentifier="dev-db",
    AllocatedStorage=100,
    ApplyImmediately=True,
)

Route 53

List Hosted Zones

route53 = boto3.client("route53")

for zone in route53.list_hosted_zones()["HostedZones"]:
    print(zone["Name"], zone["Id"])

List Records

records = route53.list_resource_record_sets(
    HostedZoneId="/hostedzone/Z1234567890",
)

for record in records["ResourceRecordSets"]:
    print(record["Name"], record["Type"])

Upsert DNS Record

route53.change_resource_record_sets(
    HostedZoneId="/hostedzone/Z1234567890",
    ChangeBatch={
        "Changes": [
            {
                "Action": "UPSERT",
                "ResourceRecordSet": {
                    "Name": "app.example.com.",
                    "Type": "A",
                    "TTL": 300,
                    "ResourceRecords": [{"Value": "203.0.113.10"}],
                },
            }
        ]
    },
)

Common Patterns

Idempotent Create Bucket

def ensure_bucket(name: str, region: str = "us-east-1") -> None:
    s3 = boto3.client("s3", region_name=region)

    try:
        s3.head_bucket(Bucket=name)
        return
    except ClientError as exc:
        status = exc.response["ResponseMetadata"]["HTTPStatusCode"]
        if status not in (403, 404):
            raise

    kwargs = {"Bucket": name}
    if region != "us-east-1":
        kwargs["CreateBucketConfiguration"] = {"LocationConstraint": region}

    s3.create_bucket(**kwargs)

Poll Until Status

def wait_for_instance_state(instance_id: str, state: str, timeout: int = 300) -> None:
    ec2 = boto3.client("ec2")
    deadline = time.time() + timeout

    while time.time() < deadline:
        response = ec2.describe_instances(InstanceIds=[instance_id])
        current = response["Reservations"][0]["Instances"][0]["State"]["Name"]
        if current == state:
            return
        time.sleep(5)

    raise TimeoutError(f"{instance_id} did not reach {state}")

Tag Resources

ec2.create_tags(
    Resources=["i-0123456789abcdef0"],
    Tags=[
        {"Key": "Environment", "Value": "prod"},
        {"Key": "Owner", "Value": "platform"},
    ],
)

Use Environment Variables

import os

session = boto3.Session(
    profile_name=os.getenv("AWS_PROFILE"),
    region_name=os.getenv("AWS_REGION", "us-east-1"),
)

Build a Reusable Client Factory

def client(service: str, profile: str | None = None, region: str = "us-east-1"):
    session = boto3.Session(profile_name=profile, region_name=region)
    return session.client(service)

s3 = client("s3", profile="prod")

Testing & Local Development

Stub AWS Calls

import boto3
from botocore.stub import Stubber

s3 = boto3.client("s3")

with Stubber(s3) as stubber:
    stubber.add_response(
        "list_buckets",
        {"Buckets": [], "Owner": {"DisplayName": "me", "ID": "123"}},
    )

    response = s3.list_buckets()

Use moto for Unit Tests

pip install moto pytest

import boto3
from moto import mock_aws

@mock_aws
def test_create_bucket():
    s3 = boto3.client("s3", region_name="us-east-1")
    s3.create_bucket(Bucket="test-bucket")

    buckets = s3.list_buckets()["Buckets"]
    assert buckets[0]["Name"] == "test-bucket"

LocalStack Endpoint

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

Production Checklist

Use IAM roles instead of hardcoded credentials.
Use least-privilege IAM policies.
Set retry and timeout configuration for automation.
Use paginators for list operations.
Use waiters for supported async workflows.
Handle ClientError explicitly.
Avoid scan on large DynamoDB tables unless you really need it.
Encrypt S3, RDS, SQS, SNS, SSM, and Secrets Manager data where required.
Tag resources consistently.
Log AWS request IDs when debugging production failures.