Talking to a Cloud SDK
Every cloud SDK is the same shape: a client built from ambient credentials, paginators, and a retry policy you set once.
s3 = boto3.client("s3", config=Config(retries={"total_max_attempts": 10, "mode": "standard"}))Every cloud SDK has the same three parts, whichever provider it is for: a client object that owns credentials, a region and a connection pool; operations that map one-to-one onto HTTP calls; and a helper that walks a paged result. Learn the shape once and the next SDK is mostly naming.
Build the client once
boto3.client("s3", region_name="eu-west-1") resolves credentials in order: the environment (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN, AWS_REGION), then the shared config file, then the instance or pod role. That ordering is why production code contains no keys — the identical line works on a laptop with a profile and in a pod with a role. Keep the client at module scope; building one per call throws away the pooled connections and re-signs from scratch.
Pagination, long form and short
The long form is a while loop that feeds the previous response's NextContinuationToken back into the next request until it is absent. The short form is client.get_paginator("list_objects_v2").paginate(Bucket=...), which yields the same pages. They do the same work; the paginator simply cannot forget the token or stop after the first thousand keys.
Throttling
A throttled call is normal, not exceptional. botocore has three retry modes: legacy (the default, up to 5 attempts), standard (3 attempts and a wider set of retryable errors) and the experimental adaptive, which rate-limits the client itself. Configure it once on the client with Config(retries={"total_max_attempts": 10, "mode": "standard"}), or per environment with AWS_RETRY_MODE and AWS_MAX_ATTEMPTS, rather than hand-rolling a loop at every call site.
Example
When to use it
- List a bucket with a million objects without writing the continuation-token loop by hand.
- Run the same code on a laptop with a profile and in a pod with an IAM role, with no key in the repo.
- Survive a throttled API during a nightly batch by setting one retry policy on the client.
- Unit test a module that writes to object storage without credentials or a network.
More examples
The real calls: manual loop against paginator
Both generators make the same API calls; the second cannot be written wrong. Note get("Contents", []) in both: an empty prefix returns a page with no Contents key at all, which is the most common KeyError in code that lists a bucket.
import boto3
from botocore.config import Config
# One client per process. Credentials come from the environment or the pod role;
# there is no key in this file and there never should be.
s3 = boto3.client(
"s3",
region_name="eu-west-1",
config=Config(retries={"total_max_attempts": 10, "mode": "standard"}),
)
# Longhand. list_objects_v2 returns at most 1000 keys per call; IsTruncated and
# NextContinuationToken are how you get the rest.
def all_keys_manual(bucket, prefix):
token = None
while True:
kwargs = {"Bucket": bucket, "Prefix": prefix}
if token:
kwargs["ContinuationToken"] = token
page = s3.list_objects_v2(**kwargs)
yield from (obj["Key"] for obj in page.get("Contents", []))
if not page.get("IsTruncated"):
return
token = page["NextContinuationToken"]
# Shorthand. Identical requests, identical results, no token bookkeeping.
def all_keys(bucket, prefix):
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
yield from (obj["Key"] for obj in page.get("Contents", []))Inject the client so tests need no network
Taking the client as a constructor argument is what makes the module testable without credentials or a network. Declaring a Protocol with just the methods you call documents the surface you depend on, so a reviewer can see that this class cannot delete anything.
from typing import Protocol
class ObjectStore(Protocol):
"""Only the operations this module actually uses."""
def put_object(self, *, Bucket: str, Key: str, Body: bytes) -> dict: ...
class ReportWriter:
def __init__(self, store: ObjectStore, bucket: str):
self.store = store # boto3 client in production, a fake in tests
self.bucket = bucket
def write(self, day: str, csv: bytes) -> str:
key = f"reports/{day}.csv"
self.store.put_object(Bucket=self.bucket, Key=key, Body=csv)
return key
# tests/test_reports.py
class FakeStore:
def __init__(self):
self.objects = {}
def put_object(self, *, Bucket, Key, Body):
self.objects[(Bucket, Key)] = Body
return {"ETag": '"stub"'}
def test_write_uses_a_dated_key():
store = FakeStore()
key = ReportWriter(store, "analytics").write("2026-10-02", b"a,b\n1,2\n")
assert key == "reports/2026-10-02.csv"
assert ("analytics", key) in store.objectsRetrying only what is safe to retry
Reach for this only when the SDK's own retry policy is not enough, and only for operations where a duplicate request is harmless. The jitter is not a detail: without it, a retry storm from every worker arrives at the same instant and throttles you again.
import random
import time
import botocore.exceptions
# The SDK retries throttling and 5xx for you. You add a layer only for errors
# it cannot know about - and only around an operation that is idempotent.
TRANSIENT = {"ThrottlingException", "TooManyRequestsException", "RequestTimeout"}
def retry_idempotent(call, *, attempts=5, base=0.2):
for n in range(attempts):
try:
return call()
except botocore.exceptions.ClientError as exc:
code = exc.response["Error"]["Code"]
if code not in TRANSIENT or n == attempts - 1:
raise
# Full jitter: sleep a random amount up to the exponential ceiling,
# so a thousand workers do not retry in lockstep.
time.sleep(random.uniform(0, base * 2 ** n))
# Safe: writing the same key with the same body twice has one outcome.
retry_idempotent(lambda: s3.put_object(Bucket="analytics", Key="x.csv", Body=b"1"))
# NOT safe to wrap blindly: a timed-out SendMessage may already have been
# accepted. Pass a deduplication id, or let the consumer dedupe on a key.
Discussion