Usage

Async/Sync Usage¶

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():
    async with AsyncS3Client() as s3:
        response = await s3.create_bucket("capo")
        print(response)

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

from capo_s3 import S3Client

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

Input/Output Types¶

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:
    response = await s3_client.create_bucket(
        bucket="some_bucket",
        create_bucket_configuration={
            "location": {
                "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¶

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

from capo_s3 import AsyncS3Client
from capo_s3.errors import NoSuchUpload


async with AsyncS3Client() as s3:
    try:
        await s3.abort_multipart_upload()
    except NoSuchUpload as e:
        print(f"Error: {e}")
        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.