capo (caporegime) - capo

Skip to content

Async/Sync Usage

Input/Output Types

Error Handling

Streaming

Waiters

Pagination

Presigning

Interceptors

Configuring the HTTP client

Retrying

Async/Sync Usage

Input/Output Types

Error Handling

Streaming

Waiters

Pagination

Presigning

Interceptors

Configuring the HTTP client

Retrying

capo (caporegime)¶

Community-driven AWS SDK for Python. Manage AWS services like a capo.

Features¶

async and sync — Use the same API for both async and sync code, with support for asyncio and trio.

WASM support — Runs in WASM environments via Pyodide.

typed and documented — Fully typed and documented for a better developer experience.

simple inputs — No need to import separate parameter classes; nested inputs are fully typed via TypedDicts.

generated from Smithy models — The SDK is generated from the official Smithy models describing AWS APIs, ensuring accuracy and consistency.

zero runtime overhead — Codegen produces dedicated serialization and deserialization code for each operation, avoiding reflection.

interchangeable input/output — Input and output types use the same TypedDicts, so you can pass a response directly as input when appropriate.

built on zapros — A modern HTTP client for Python that abstracts HTTP semantics from the transport implementation.

fast to import — import time stays flat even for huge services like EC2.

interceptors — Hook into the operation request/response lifecycle to inspect, log, or modify calls.

Warning<br>The API should be mostly stable, but some breaking changes may occur as the SDK is still in early development. We strongly recommend pinning the version before the first major release.

Installation&para;

Every service is a standalone package, so you only install the ones you actually use. Packages are published under the capo- prefix:

uv add capo-ec2 # Amazon EC2 (also covers Amazon VPC)<br>uv add capo-s3 # Amazon S3<br>uv add capo-iam # AWS IAM<br>uv add capo-rds # Amazon RDS<br>uv add capo-lambda # AWS Lambda<br>uv add capo-cloudwatch # Amazon CloudWatch<br>uv add capo-elastic-load-balancing # Elastic Load Balancing (ELB)<br>uv add capo-route-53 # Amazon Route 53<br>uv add capo-cloudfront # Amazon CloudFront

Note that Amazon VPC has no separate package — its API is part of EC2, so capo-ec2 covers it.

Services not yet on PyPI&para;

Not every service is on PyPI yet. We publish incrementally because of PyPI's limits on new projects and total upload size. If the service you need isn't published, install it straight from the repository:

uv add git+https://github.com/kap-sh/capo#subdirectory=services/glacier

Replace glacier with the directory name of the service under services/, and please open an issue so we can prioritize publishing it.

Async/Sync Usage&para;

All the services have both async and sync client. The async client is simply prefixed with Async (e.g. AsyncS3Client).

from capo_s3 import AsyncS3Client

async def main():<br>async with AsyncS3Client() as s3:<br>response = await s3.create_bucket("capo")<br>print(response)

Sync usage is the same, just without the Async prefix and without await:

from capo_s3 import S3Client

with S3Client() as s3:<br>response = s3.create_bucket("capo")<br>print(response)

Input/Output Types&para;

No matter how nested the input types are, you don't need to import any additional classes. All input and output types are fully typed via TypedDicts.

from capo_s3 import AsyncS3Client

async with AsyncS3Client() as s3_client:<br>response = await s3_client.create_bucket(<br>bucket="some_bucket",<br>create_bucket_configuration={<br>"location": {<br>"name": "location_name"

print(response["location"])

The output types are also TypeDicts, we do this to make the input/output types interchangeable. You can pass the output member to another operation that accepts the same type as input.

Error Handling&para;

The SDK raises exceptions for errors returned by the API. Catch them to handle failures gracefully.

from capo_s3 import AsyncS3Client<br>from capo_s3.errors import NoSuchUpload

async with AsyncS3Client() as s3:<br>try:<br>await s3.abort_multipart_upload()<br>except NoSuchUpload as e:<br>print(f"Error: {e}")<br>print(e.data) # additional error data

Note that the service errors (errors returned by AWS) might have additional data of any shape stored in the data attribute, which is a TypedDict. You can access it to get more information about the error.

All the service errors that an operation can raise are documented in that operation method's docstring.

Streaming&para;

If the operation's input is a streaming blob, you can pass any AsyncIterator[bytes] or just a bytes object.

from capo_s3 import AsyncS3Client

s3_client = AsyncS3Client()

response = await s3_client.put_object("bucket_name", "key", body=b"some binary data")

Or, if you don't want to load the entire blob into memory, you can pass an AsyncIterator[bytes]:

from capo_s3 import AsyncS3Client

async def async_iterator():<br>yield b"capo"

response = await...