aioboto3 Documentation

The usage of aioboto3 closely mimics boto3, with the primary difference being that most operations must be prefixed with await.

Key differences from boto3:

• aioboto3.resource returns a resource object that is an async context manager (supports async with) and has an awaitable .close() method.
• Service resources (e.g., s3.Bucket, dynamodb.Table) must be instantiated using await.
• Standard client connections also work using the async with session.client(...) pattern.

import asyncio
import aioboto3
from boto3.dynamodb.conditions import Key

async def main():
    session = aioboto3.Session()
    # Use async with for resources
    async with session.resource('dynamodb', region_name='eu-central-1') as dynamo_resource:
        # Service resources must be awaited
        table = await dynamo_resource.Table('test_table')

        await table.put_item(
            Item={'pk': 'test1', 'col1': 'some_data'}
        )

        result = await table.query(
            KeyConditionExpression=Key('pk').eq('test1')
        )
        print(result['Items'])

asyncio.run(main())

AWS S3 Client-Side Encryption (CSE) allows you to encrypt data before it is sent to S3, ensuring that sensitive information is never unencrypted in transit or at rest on the S3 server.

aioboto3 provides the S3CSE class and several CryptoContext implementations to manage this process. You wrap your S3 operations (like put_object and get_object) using an S3CSE instance initialized with a specific CryptoContext.

import asyncio
import aioboto3
from aioboto3.s3.cse import S3CSE, KMSCryptoContext

async def main():
    # 1. Initialize the CryptoContext (e.g., KMS)
    ctx = KMSCryptoContext(keyid='alias/someKey', kms_client_args={'region_name': 'eu-central-1'})

    some_data = b'Some sensitive data for S3'

    # 2. Use S3CSE as an async context manager
    async with S3CSE(crypto_context=ctx, s3_client_args={'region_name': 'eu-central-1'}) as s3_cse:
        # 3. Perform encrypted uploads
        await s3_cse.put_object(
            Body=some_data,
            Bucket='some-bucket',
            Key='encrypted_file',
        )

        # 4. Perform encrypted downloads
        response = await s3_cse.get_object(
            Bucket='some-bucket',
            Key='encrypted_file'
        )
        data = await response['Body'].read()
        print(data)

asyncio.run(main())

You can use aioboto3.experimental.async_chalice.AsyncChalice as the main application entrypoint for an AWS Chalice app. This integration provides shims that allow you to use async def functions for HTTP routes.

Key features:

• Async Routes: Define routes using async def to handle asynchronous logic.
• aioboto3 Session: The app.aioboto3 attribute contains an aioboto3.Session object, which you can use to create clients (e.g., app.aioboto3.client('s3')).
• Custom Sessions: You can pass a custom session to AsyncChalice to override the default empty session.

Warning: This integration is EXPERIMENTAL. It is not recommended for critical production use. Due to how Chalice manages the event loop (which can disappear between invocations), you should not cache aioboto3 clients or resources.

from aioboto3.experimental.async_chalice import AsyncChalice

# Initialize the app
app = AsyncChalice(app_name='testclient')

# Define an async route
@app.route('/hello/{name}')
async def hello(name):
    return {'hello': name}

# Use the integrated aioboto3 session to create clients
@app.route('/list_buckets')
async def get_list_buckets():
    async with app.aioboto3.client("s3") as s3:
        resp = await s3.list_buckets()

    return {"buckets": [bucket['Name'] for bucket in resp['Buckets']]}

Since aioboto3 v8.0.0, .client and .resource are async context managers. In long-running applications like aiohttp servers, you should use contextlib.AsyncExitStack to manage the lifecycle of these resources during application startup and shutdown to avoid unclosed session warnings.

import contextlib
import aioboto3
from aiohttp import web

session = aioboto3.Session()
routes = web.RouteTableDef()

async def startup_tasks(app: web.Application) -> None:
    context_stack = contextlib.AsyncExitStack()
    app['context_stack'] = context_stack

    # Enter the resource context and save it in the stack
    app['dynamo_resource'] = await context_stack.enter_async_context(
        session.resource('dynamodb', region_name='eu-west-1')
    )
    # Await the service resource
    app['table'] = await app['dynamo_resource'].Table('my_table')

async def shutdown_tasks(app: web.Application) -> None:
    # Clean up all managed contexts
    await app['context_stack'].aclose()

_app = web.Application()
_app.on_startup.append(startup_tasks)
_app.on_shutdown.append(shutdown_tasks)