Python Boto3 Cheatsheet Python Boto3 Cheatsheet

python -m pip install boto3

import boto3

# Client - low-level, 1:1 with the AWS API. Prefer this; it covers every service.
s3 = boto3.client("s3", region_name="us-east-1")

# Resource - higher-level OO wrapper (S3, EC2, DynamoDB, and a few others only;
# in maintenance mode - no new services get resources)
s3r = boto3.resource("s3")

# Session - explicit credentials/profile/region; use for multi-account scripts
session = boto3.Session(profile_name="prod", region_name="eu-west-1")
ec2 = session.client("ec2")

# Cross-account via STS assume-role
sts = boto3.client("sts")
creds = sts.assume_role(
    RoleArn="arn:aws:iam::123456789012:role/Automation",
    RoleSessionName="audit",
)["Credentials"]
target = boto3.client(
    "s3",
    aws_access_key_id=creds["AccessKeyId"],
    aws_secret_access_key=creds["SecretAccessKey"],
    aws_session_token=creds["SessionToken"],
)

s3 = boto3.client("s3")

# Upload / download (managed transfers - multipart automatically for big files)
s3.upload_file("build.tar.gz", "my-bucket", "releases/build.tar.gz",
               ExtraArgs={"ServerSideEncryption": "aws:kms"})
s3.download_file("my-bucket", "releases/build.tar.gz", "/tmp/build.tar.gz")

# Small objects in memory
s3.put_object(Bucket="my-bucket", Key="cfg.json", Body=b'{"env":"prod"}')
body = s3.get_object(Bucket="my-bucket", Key="cfg.json")["Body"].read()

# Presigned URL - temporary access without credentials
url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "my-bucket", "Key": "report.pdf"},
    ExpiresIn=3600,          # seconds
)
# For uploads: "put_object", or generate_presigned_post for browser forms

# List with pagination - list_objects_v2 caps at 1000, ALWAYS paginate
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket="my-bucket", Prefix="logs/2026/"):
    for obj in page.get("Contents", []):
        print(obj["Key"], obj["Size"])

s3.delete_object(Bucket="my-bucket", Key="old.log")
s3.copy_object(Bucket="my-bucket", Key="dst",
               CopySource={"Bucket": "my-bucket", "Key": "src"})

ec2 = boto3.client("ec2")

# Describe with filters
resp = ec2.describe_instances(Filters=[
    {"Name": "tag:Env", "Values": ["prod"]},
    {"Name": "instance-state-name", "Values": ["running"]},
])
for reservation in resp["Reservations"]:
    for inst in reservation["Instances"]:
        name = next((t["Value"] for t in inst.get("Tags", [])
                     if t["Key"] == "Name"), "-")
        print(inst["InstanceId"], inst["InstanceType"],
              inst.get("PrivateIpAddress"), name)

# Start / stop / reboot / terminate
ec2.stop_instances(InstanceIds=["i-0abc123"])
ec2.start_instances(InstanceIds=["i-0abc123"])
ec2.terminate_instances(InstanceIds=["i-0abc123"], DryRun=True)  # permission check

# Tags
ec2.create_tags(Resources=["i-0abc123"],
                Tags=[{"Key": "Owner", "Value": "platform-team"}])
ec2.delete_tags(Resources=["i-0abc123"], Tags=[{"Key": "Temp"}])

import json, base64
lam = boto3.client("lambda")

# Synchronous invoke
resp = lam.invoke(
    FunctionName="report-gen",
    InvocationType="RequestResponse",     # "Event" = async fire-and-forget
    Payload=json.dumps({"day": "2026-07-23"}),
    LogType="Tail",                       # last 4KB of logs in the response
)
result = json.loads(resp["Payload"].read())
logs = base64.b64decode(resp["LogResult"]).decode()
if resp.get("FunctionError"):             # function raised - payload holds the error
    raise RuntimeError(result)

# Deploy new code from a zip
with open("function.zip", "rb") as f:
    lam.update_function_code(FunctionName="report-gen", ZipFile=f.read())
lam.get_waiter("function_updated_v2").wait(FunctionName="report-gen")

lam.update_function_configuration(FunctionName="report-gen",
                                  MemorySize=512, Timeout=30,
                                  Environment={"Variables": {"ENV": "prod"}})

from boto3.dynamodb.conditions import Key, Attr

table = boto3.resource("dynamodb").Table("deployments")   # resource handles types for you

table.put_item(Item={
    "service": "api",              # partition key
    "deployed_at": "2026-07-23T10:00:00Z",   # sort key
    "version": "1.4.2",
    "healthy": True,
})

item = table.get_item(Key={"service": "api",
                           "deployed_at": "2026-07-23T10:00:00Z"}).get("Item")

# Query - partition key required, sort key optional
resp = table.query(
    KeyConditionExpression=Key("service").eq("api")
                           & Key("deployed_at").begins_with("2026-07"),
    FilterExpression=Attr("healthy").eq(True),
    ScanIndexForward=False,        # newest first
    Limit=10,
)
items = resp["Items"]

# Batch write
with table.batch_writer() as batch:
    for item in items:
        batch.put_item(Item=item)

ssm = boto3.client("ssm")

ssm.put_parameter(Name="/prod/api/db-url", Value="postgres://...",
                  Type="SecureString", Overwrite=True)

val = ssm.get_parameter(Name="/prod/api/db-url",
                        WithDecryption=True)["Parameter"]["Value"]

# All params under a path
paginator = ssm.get_paginator("get_parameters_by_path")
for page in paginator.paginate(Path="/prod/api/", Recursive=True,
                               WithDecryption=True):
    for p in page["Parameters"]:
        print(p["Name"], "=", p["Value"])

from botocore.exceptions import ClientError, NoCredentialsError

try:
    s3.head_object(Bucket="my-bucket", Key="maybe-missing.txt")
except ClientError as e:
    code = e.response["Error"]["Code"]
    if code == "404":                          # head_object returns 404, not NoSuchKey
        print("object does not exist")
    elif code in ("AccessDenied", "403"):
        print("check IAM policy")
    else:
        raise
except NoCredentialsError:
    print("no credentials found - check the resolution chain")

# Any list/describe call with a NextToken has a paginator - never loop tokens by hand
ec2_pag = ec2.get_paginator("describe_instances")
ids = [i["InstanceId"]
       for page in ec2_pag.paginate()
       for r in page["Reservations"]
       for i in r["Instances"]]

# Waiters - poll until a state is reached, with sane backoff built in
ec2.get_waiter("instance_running").wait(
    InstanceIds=["i-0abc123"],
    WaiterConfig={"Delay": 15, "MaxAttempts": 40},
)
s3.get_waiter("object_exists").wait(Bucket="my-bucket", Key="flag.done")

from botocore.config import Config

cfg = Config(
    region_name="us-east-1",
    retries={"max_attempts": 10, "mode": "adaptive"},  # adaptive throttles client-side
    connect_timeout=5,
    read_timeout=60,
    max_pool_connections=50,       # raise when fanning out across threads
)
s3 = boto3.client("s3", config=cfg)