### Install local copy and virtual environment

Source: https://github.com/terricain/aioboto3/blob/main/docs/contributing.md

Navigate to the cloned directory and install the project within a virtual environment using uv.

```shell
cd aioboto3/
uv sync
```

--------------------------------

### Install aioboto3 from source

Source: https://github.com/terricain/aioboto3/blob/main/docs/installation.md

Install aioboto3 after downloading the source code. This method is useful for development or specific build requirements.

```bash
python setup.py install
```

--------------------------------

### Install aioboto3 using pip

Source: https://github.com/terricain/aioboto3/blob/main/docs/installation.md

Install the latest stable release of aioboto3 using pip. This is the recommended installation method.

```bash
pip install aioboto3
```

--------------------------------

### Async Chalice API Example

Source: https://context7.com/terricain/aioboto3/llms.txt

Demonstrates using AsyncChalice for asynchronous AWS API calls within a Chalice application. Requires installation with `pip install aioboto3[chalice]`.

```python
from aioboto3.experimental.async_chalice import AsyncChalice
import aioboto3

app = AsyncChalice(app_name="my-async-api")
# app.aioboto3 is an aioboto3.Session instance, set automatically


@app.route("/items/{item_id}")
async def get_item(item_id: str):
    async with app.aioboto3.resource("dynamodb", region_name="us-east-1") as dynamo:
        table = await dynamo.Table("items")
        response = await table.get_item(Key={"item_id": item_id})
        item = response.get("Item")
        if not item:
            return {"error": "not found"}, 404
        return item


@app.route("/items", methods=["POST"])
async def create_item():
    body = app.current_request.json_body
    async with app.aioboto3.resource("dynamodb", region_name="us-east-1") as dynamo:
        table = await dynamo.Table("items")
        await table.put_item(Item=body)
    return {"status": "created"}
```

--------------------------------

### Download aioboto3 tarball

Source: https://github.com/terricain/aioboto3/blob/main/docs/installation.md

Download the source tarball for aioboto3 from GitHub to install from source.

```bash
curl  -OL https://github.com/terrycain/aioboto3/tarball/master
```

--------------------------------

### Retry Attempts Output Example

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Example output demonstrating the number of retries required for concurrent AWS Organizations ListRoots API calls when custom retry configurations are applied.

```python3
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 5, 4, 6, 4, 4, 4, 4, 4, 5, 5]
```

--------------------------------

### Asymmetric Encryption Example

Source: https://github.com/terricain/aioboto3/blob/main/resources/S3-CSE/README.md

Encrypt S3 objects using asymmetric encryption. This requires a directory containing the encryption keys.

```bash
java -jar build/libs/s3cse-1.0.jar --crypto-type asymmetric --bucket-name bucket1 --key-name test-cse-asymmetric \
                                   --region eu-west-1 --key-dir ./keys
```

--------------------------------

### KMS Encryption Example

Source: https://github.com/terricain/aioboto3/blob/main/resources/S3-CSE/README.md

Use this command for S3 object encryption with AWS KMS. Requires a KMS key ID and the --authenticated-crypto flag.

```bash
java -jar build/libs/s3cse-1.0.jar --crypto-type kms --bucket-name bucket1 --key-name test-cse-kms \
                                   --region eu-west-1 --kms-key-id alias/someKey --authenticated-crypto
```

--------------------------------

### DynamoDB Item Put and Query Example

Source: https://github.com/terricain/aioboto3/blob/main/docs/readme.md

Shows how to interact with a DynamoDB table asynchronously using aioboto3. Includes putting an item, querying by a key condition, and performing a batch write operation. Ensure the DynamoDB table and region are correctly configured.

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


async def main():
    session = aioboto3.Session()
    async with session.resource('dynamodb', region_name='eu-central-1') as dynamo_resource:
        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')
        )

        # Example batch write
        more_items = [{'pk': 't2', 'col1': 'c1'}, \
                      {'pk': 't3', 'col1': 'c3'}]
        async with table.batch_writer() as batch:
            for item_ in more_items:
                await batch.put_item(Item=item_)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
```

--------------------------------

### AioHTTP Server with AsyncExitStack for DynamoDB Resource

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Provides an example of integrating aioboto3 resources into an aiohttp web server using AsyncExitStack to manage resource lifecycle and ensure proper cleanup.

```python
"
contextlib.AsyncExitStack requires python 3.7
"import contextlib

import aioboto3
from boto3.dynamodb.conditions import Key
from aiohttp import web

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


@routes.get('/')
async def hello(request):

    # request.app['table'] == Table object from boto3 docs
    response = await request.app['table'].query(
        KeyConditionExpression=Key('id').eq('lalalala')
    )

    return web.Response(text=str(response))


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

    app['dynamo_resource'] = await context_stack.enter_async_context(
        session.resource('dynamodb', region_name='eu-west-1')
    )
    # By now, app['dynamo_resource'] will have methods like .Table() and list_tables() etc...

    # aioboto3 v8.0.0+ all service resources (aka Table(), Bucket() etc...) need to be awaited
    app['table'] = await app['dynamo_resource'].Table('somedynamodbtablename')


async def shutdown_tasks(app: web.Application) -> None:
    await app['context_stack'].aclose()
    # By now, app['dynamo_resource'] would be closed


_app = web.Application()
_app.add_routes(routes)
_app.on_startup.append(startup_tasks)
_app.on_shutdown.append(shutdown_tasks)
web.run_app(_app, port=8000)
```

--------------------------------

### Chalice App with Async Routes and S3 Client

Source: https://github.com/terricain/aioboto3/blob/main/docs/chalice.md

This example demonstrates setting up an AWS Chalice application using AsyncChalice for asynchronous HTTP routes. It shows how to define a route that interacts with AWS S3 using an aioboto3 client obtained from the app's session.

```python
from aioboto3.experimental.async_chalice import AsyncChalice

app = AsyncChalice(app_name='testclient')


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


@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']]}

```

--------------------------------

### Clone aioboto3 repository

Source: https://github.com/terricain/aioboto3/blob/main/docs/installation.md

Clone the public repository of aioboto3 from GitHub to install from source.

```bash
git clone git://github.com/terrycain/aioboto3
```

--------------------------------

### Symmetric Encryption Example

Source: https://github.com/terricain/aioboto3/blob/main/resources/S3-CSE/README.md

Encrypt S3 objects using symmetric encryption. This requires a directory containing the encryption keys.

```bash
java -jar build/libs/s3cse-1.0.jar --crypto-type symmetric --bucket-name bucket1 --key-name test-cse-symmetric \
                                   --region eu-west-1 --key-dir ./keys
```

--------------------------------

### DynamoDB Table Operations

Source: https://github.com/terricain/aioboto3/blob/main/README.rst

Example of using aioboto3 to interact with DynamoDB, including putting items, querying, and using a batch writer. Requires importing asyncio, aioboto3, and Key from boto3.dynamodb.conditions.

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


async def main():
    session = aioboto3.Session()
    async with session.resource('dynamodb', region_name='eu-central-1') as dynamo_resource:
        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')
        )

        # Example batch write
        more_items = [{'pk': 't2', 'col1': 'c1'}, \
                      {'pk': 't3', 'col1': 'c3'}]
        async with table.batch_writer() as batch:
            for item_ in more_items:
                await batch.put_item(Item=item_)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# Outputs:
#  [{'col1': 'some_data', 'pk': 'test1'}]
```

--------------------------------

### S3 Client-Side Encryption with KMS Managed Keys Example

Source: https://github.com/terricain/aioboto3/blob/main/docs/cse.md

Demonstrates how to upload and download an object to/from S3 using client-side encryption with KMS managed keys. Ensure you have a KMS key configured and appropriate IAM permissions.

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

async def main():
    ctx = KMSCryptoContext(keyid='alias/someKey', kms_client_args={'region_name': 'eu-central-1'})

    some_data = b'Some sensitive data for S3'

    async with S3CSE(crypto_context=ctx, s3_client_args={'region_name': 'eu-central-1'}) as s3_cse:
        # Upload some binary data
        await s3_cse.put_object(
            Body=some_data,
            Bucket='some-bucket',
            Key='encrypted_file',
        )

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

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
```

--------------------------------

### Configure Client Retries with AioConfig

Source: https://context7.com/terricain/aioboto3/llms.txt

Shows how to configure retry behavior for an async client using `aiobotocore.config.AioConfig`. This is useful for high-concurrency scenarios. Both DynamoDB and SSM client examples are provided.

```python
import asyncio
import aioboto3
from aiobotocore.config import AioConfig

async def main():
    session = aioboto3.Session()

    # Configure retries for high-concurrency scenarios
    config = AioConfig(retries={"max_attempts": 10, "mode": "adaptive"})

    async with session.client("dynamodb", region_name="eu-west-1", config=config) as dynamo:
        # List all tables
        response = await dynamo.list_tables()
        print(response["TableNames"])

    # SSM Parameter Store example
    async with session.client("ssm", region_name="eu-central-1") as ssm:
        result = await ssm.get_parameter(Name="/myapp/db_password", WithDecryption=True)
        print(result["Parameter"]["Value"])

asyncio.run(main())
```

--------------------------------

### Interact with DynamoDB Table Resources

Source: https://context7.com/terricain/aioboto3/llms.txt

Demonstrates using a high-level DynamoDB resource object to perform common operations like putting, getting, querying, and scanning items. Requires `boto3.dynamodb.conditions` for expressions.

```python
import asyncio
import aioboto3
from boto3.dynamodb.conditions import Key, Attr

async def main():
    session = aioboto3.Session()

    async with session.resource("dynamodb", region_name="us-east-1") as dynamo:
        table = await dynamo.Table("users")

        # Put an item
        await table.put_item(Item={"user_id": "u-001", "name": "Alice", "age": 30})

        # Get an item
        response = await table.get_item(Key={"user_id": "u-001"})
        print(response["Item"])

        # Query with condition expression
        result = await table.query(
            KeyConditionExpression=Key("user_id").eq("u-001")
        )
        print(result["Items"])

        # Scan with filter
        result = await table.scan(FilterExpression=Attr("age").gte(18))
        print(result["Count"])

asyncio.run(main())
```

--------------------------------

### S3 Client-Side Encryption with KMS

Source: https://context7.com/terricain/aioboto3/llms.txt

Demonstrates transparently encrypting and decrypting S3 objects using `S3CSE` with KMS-managed keys. Supports authenticated encryption (AES-GCM) for range gets.

```python
import asyncio
import aioboto3
from aioboto3.s3.cse import S3CSE, KMSCryptoContext, SymmetricCryptoContext, AsymmetricCryptoContext
from cryptography.hazmat.primitives.serialization import load_pem_private_key, load_pem_public_key

async def main():
    # --- KMS-managed keys (recommended) ---
    kms_ctx = KMSCryptoContext(
        keyid="alias/my-s3-key",
        kms_client_args={"region_name": "us-east-1"},
        authenticated_encryption=True,  # AES-GCM (supports range gets)
    )
    async with S3CSE(crypto_context=kms_ctx, s3_client_args={"region_name": "us-east-1"}) as s3_cse:
        await s3_cse.put_object(
            Body=b"Sensitive payload",
            Bucket="encrypted-bucket",
            Key="secret/data.bin",
        )
        response = await s3_cse.get_object(Bucket="encrypted-bucket", Key="secret/data.bin")
        plaintext = await response["Body"].read()
        print(plaintext)  # b"Sensitive payload"

        # Range-get on AES-GCM encrypted object
        ranged = await s3_cse.get_object(
            Bucket="encrypted-bucket", Key="secret/data.bin", Range="bytes=0-6"
        )
        print(await ranged["Body"].read())  # b"Sensiti"

    # --- Symmetric key ---
    sym_ctx = SymmetricCryptoContext(key=b"\x00" * 32)  # 256-bit AES key
    async with S3CSE(crypto_context=sym_ctx) as s3_cse:
        await s3_cse.put_object(Body=b"Hello", Bucket="my-bucket", Key="sym_encrypted.bin")

asyncio.run(main())
```

--------------------------------

### DynamoDB: Batch Write Items

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Utilizes the batch writer for efficient DynamoDB item writing, handling retries automatically. This example shows writing multiple items within a single batch operation.

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


async def main():
    session = aioboto3.Session()
    async with session.resource('dynamodb', region_name='eu-central-1') as dynamo_resource:
        table = await dynamo_resource.Table('test_table')

        # As the default batch size is 25, all of these will be written in one batch
        async with table.batch_writer() as dynamo_writer:
            await dynamo_writer.put_item(Item={'pk': 'test1', 'col1': 'some_data'})
            await dynamo_writer.put_item(Item={'pk': 'test2', 'col1': 'some_data'})
            await dynamo_writer.put_item(Item={'pk': 'test3', 'col1': 'some_data'})
            await dynamo_writer.put_item(Item={'pk': 'test4', 'col1': 'some_data'})
            await dynamo_writer.put_item(Item={'pk': 'test5', 'col1': 'some_data'})

        result = await table.scan()

        print(result['Count'])

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# Outputs:
#  5
```

--------------------------------

### Initialize aioboto3 Session and Use Clients/Resources

Source: https://context7.com/terricain/aioboto3/llms.txt

Demonstrates initializing an aioboto3 session with explicit credentials and using both low-level clients (STS) and high-level resources (S3) as async context managers.

```python
import asyncio
import aioboto3

async def main():
    # Explicit credentials (or omit to use environment/IAM role)
    session = aioboto3.Session(
        aws_access_key_id="AKIAIOSFODNN7EXAMPLE",
        aws_secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
        region_name="us-east-1",
    )

    # Low-level client usage
    async with session.client("sts") as sts:
        identity = await sts.get_caller_identity()
        print(identity["Account"])

    # High-level resource usage
    async with session.resource("s3") as s3:
        bucket = await s3.Bucket("my-bucket")
        print(bucket.name)

asyncio.run(main())
```

--------------------------------

### aioboto3.Session Usage

Source: https://context7.com/terricain/aioboto3/llms.txt

Demonstrates how to create an asynchronous session and use both low-level clients and high-level resources.

```APIDOC
## aioboto3.Session

The entry point for all aioboto3 usage. Replaces `boto3.Session` with an async-capable session that creates aiobotocore-backed clients and resource objects. Credentials, region, and profile can be passed directly or resolved from the environment/config files. Both `.client()` and `.resource()` return async context managers.

```python
import asyncio
import aioboto3

async def main():
    # Explicit credentials (or omit to use environment/IAM role)
    session = aioboto3.Session(
        aws_access_key_id="AKIAIOSFODNN7EXAMPLE",
        aws_secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
        region_name="us-east-1",
    )

    # Low-level client usage
    async with session.client("sts") as sts:
        identity = await sts.get_caller_identity()
        print(identity["Account"])

    # High-level resource usage
    async with session.resource("s3") as s3:
        bucket = await s3.Bucket("my-bucket")
        print(bucket.name)

asyncio.run(main())
```
```

--------------------------------

### Describe SSM Parameters with aioboto3 Client

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Shows how to establish an asynchronous client connection to AWS Systems Manager (SSM) and retrieve parameter descriptions using aioboto3.

```python3
import asyncio
import aioboto3


async def main():
    session = aioboto3.Session()
    async with session.client('ssm', region_name='eu-central-1') as ssm_client:
        result = await ssm_client.describe_parameters()

        print(result['Parameters'])


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
```

--------------------------------

### Display JAR Options

Source: https://github.com/terricain/aioboto3/blob/main/resources/S3-CSE/README.md

Run this command to view the available options for the S3 CSE JAR.

```bash
java -jar build/libs/s3cse-1.0.jar -h
```

--------------------------------

### DynamoDB: Put and Query Item

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Demonstrates putting an item into a DynamoDB table and then querying it using the Key().eq() abstraction. Requires an active DynamoDB table and aioboto3 session.

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


async def main():
    session = aioboto3.Session()
    async with session.resource('dynamodb', region_name='eu-central-1') as dynamo_resource:
        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'])

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# Outputs:
#  [{'col1': 'some_data', 'pk': 'test1'}]
```

--------------------------------

### Clone aioboto3 repository

Source: https://github.com/terricain/aioboto3/blob/main/CONTRIBUTING.rst

Clone your forked repository locally to begin development.

```shell
git clone git@github.com:your_name_here/aioboto3.git
```

--------------------------------

### Session.client(service_name, ...)

Source: https://context7.com/terricain/aioboto3/llms.txt

Creates a low-level async service client for any AWS service. Must be used as an async context manager and accepts parameters similar to `boto3.Session.client()`, plus `AioConfig` for async-specific configurations.

```APIDOC
## Session.client(service_name, ...)

Creates a low-level async service client for any AWS service supported by botocore. Must be used as an async context manager. Accepts the same parameters as `boto3.Session.client()`, plus `AioConfig` for async-specific configuration such as retry behaviour.

```python
import asyncio
import aioboto3
from aiobotocore.config import AioConfig

async def main():
    session = aioboto3.Session()

    # Configure retries for high-concurrency scenarios
    config = AioConfig(retries={"max_attempts": 10, "mode": "adaptive"})

    async with session.client("dynamodb", region_name="eu-west-1", config=config) as dynamo:
        # List all tables
        response = await dynamo.list_tables()
        print(response["TableNames"])

    # SSM Parameter Store example
    async with session.client("ssm", region_name="eu-central-1") as ssm:
        result = await ssm.get_parameter(Name="/myapp/db_password", WithDecryption=True)
        print(result["Parameter"]["Value"])

asyncio.run(main())
```
```

--------------------------------

### Async S3 Bucket Object Iteration

Source: https://github.com/terricain/aioboto3/blob/main/docs/readme.md

Demonstrates how to asynchronously iterate over objects in an S3 bucket using a session resource. Requires creating a session and using the resource as an async context manager.

```python
async def main():
    session = aioboto3.Session()
    async with session.resource("s3") as s3:
        bucket = await s3.Bucket('mybucket')  # <----------------
        async for s3_object in bucket.objects.all():
            print(s3_object)
```

--------------------------------

### Iterate and Delete S3 Objects with aioboto3

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Demonstrates how to iterate over S3 objects in a bucket, filter them by prefix, and delete them using aioboto3's async resource interface.

```python3
import aioboto3


async def main():
    session = aioboto3.Session()
    async with session.resource("s3") as s3:

        bucket = await s3.Bucket('mybucket')
        async for s3_object in bucket.objects.all():
            print(s3_object)

        async for s3_object in bucket.objects.filter(Prefix='someprefix/'):
            print(s3_object)

        await bucket.objects.all().delete()

        # or
        await bucket.objects.filter(Prefix='test/').delete()
```

--------------------------------

### Long-running Server with AsyncExitStack

Source: https://context7.com/terricain/aioboto3/llms.txt

Manages aioboto3 sessions for long-running processes like web servers using `contextlib.AsyncExitStack`. Sessions are opened at startup and closed cleanly at shutdown, avoiding per-request session wrapping.

```python
import contextlib
import asyncio
import aioboto3
from boto3.dynamodb.conditions import Key
from aiohttp import web

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


@routes.get("/users/{user_id}")
async def get_user(request: web.Request) -> web.Response:
    user_id = request.match_info["user_id"]
    table = request.app["users_table"]
    response = await table.get_item(Key={"user_id": user_id})
    item = response.get("Item")
    if not item:
        raise web.HTTPNotFound()
    return web.json_response(item)


async def startup(app: web.Application) -> None:
    stack = contextlib.AsyncExitStack()
    app["exit_stack"] = stack
    dynamo = await stack.enter_async_context(
        session.resource("dynamodb", region_name="us-east-1")
    )
    app["users_table"] = await dynamo.Table("users")


async def shutdown(app: web.Application) -> None:
    await app["exit_stack"].aclose()


app = web.Application()
app.add_routes(routes)
app.on_startup.append(startup)
app.on_shutdown.append(shutdown)
web.run_app(app, port=8000)
```

--------------------------------

### S3: Stream Download and Serve via aiohttp

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Downloads an S3 object in chunks and serves it via an aiohttp web response. It sets appropriate HTTP headers for the response, including content type and disposition.

```python
from aiohttp import web
from multidict import MultiDict


async def serve_blob(
    suite: str,
    release: str,
    filename: str,
    bucket: str,
    request: web.Request,
    chunk_size: int = 69 * 1024
) -> web.StreamResponse:
    blob_s3_key = f"{suite}/{release}/{filename}"

    session = aioboto3.Session()
    async with session.client("s3") as s3:
        LOG.info(f"Serving {bucket} {blob_s3_key}")
        s3_ob = await s3.get_object(Bucket=bucket, Key=blob_s3_key)

        ob_info = s3_ob["ResponseMetadata"]["HTTPHeaders"]
        resp = web.StreamResponse(
            headers=MultiDict(
                {
                    "CONTENT-DISPOSITION": (
                        f"attachment; filename='{filename}'"
                    ),
                    "Content-Type": ob_info["content-type"],
                }
            )
        )
        resp.content_type = ob_info["content-type"]
        resp.content_length = ob_info["content-length"]
        await resp.prepare(request)

        stream = s3_ob["Body"]
        while file_data := await stream.read(chunk_size):
            await resp.write(file_data)

    return resp
```

--------------------------------

### Create a new branch for development

Source: https://github.com/terricain/aioboto3/blob/main/CONTRIBUTING.rst

Create a new branch for your bugfix or feature development.

```shell
git checkout -b name-of-your-bugfix-or-feature
```

--------------------------------

### Session.resource(service_name, ...)

Source: https://context7.com/terricain/aioboto3/llms.txt

Creates a high-level async resource object for AWS services like DynamoDB or S3. This must also be used as an async context manager, and its sub-resources require awaiting.

```APIDOC
## Session.resource(service_name, ...)

Creates a high-level async resource object (e.g., for DynamoDB or S3). Must be used as an async context manager. Service sub-resources such as `Table` or `Bucket` must themselves be awaited after creation.

```python
import asyncio
import aioboto3
from boto3.dynamodb.conditions import Key, Attr

async def main():
    session = aioboto3.Session()

    async with session.resource("dynamodb", region_name="us-east-1") as dynamo:
        table = await dynamo.Table("users")

        # Put an item
        await table.put_item(Item={"user_id": "u-001", "name": "Alice", "age": 30})

        # Get an item
        response = await table.get_item(Key={"user_id": "u-001"})
        print(response["Item"])

        # Query with condition expression
        result = await table.query(
            KeyConditionExpression=Key("user_id").eq("u-001")
        )
        print(result["Items"])

        # Scan with filter
        result = await table.scan(FilterExpression=Attr("age").gte(18))
        print(result["Count"])

asyncio.run(main())
```
```

--------------------------------

### Manage S3 bucket objects using aioboto3 resource

Source: https://context7.com/terricain/aioboto3/llms.txt

Provides async iteration and awaitable methods for S3 bucket objects, including listing all objects, filtering by prefix, and deleting objects. Object metadata like size is available as coroutine properties.

```python
import asyncio
import aioboto3

async def main():
    session = aioboto3.Session()

    async with session.resource("s3", region_name="us-east-1") as s3:
        bucket = await s3.Bucket("my-bucket")

        # List all objects
        async for obj in bucket.objects.all():
            print(obj.key, await obj.size)  # "path/file.txt", 1024

        # Filter by prefix
        async for obj in bucket.objects.filter(Prefix="logs/2025/"):
            print(obj.key)

        # Delete all objects under a prefix
        await bucket.objects.filter(Prefix="temp/").delete()

        # Delete all objects in bucket (useful before bucket deletion)
        await bucket.objects.all().delete()

asyncio.run(main())
```

--------------------------------

### Lint and test local changes

Source: https://github.com/terricain/aioboto3/blob/main/CONTRIBUTING.rst

Ensure your changes pass linting and all tests, including testing across different Python versions with tox.

```shell
make lint
```

```shell
make test
```

--------------------------------

### s3_client.download_file()

Source: https://context7.com/terricain/aioboto3/llms.txt

Downloads an S3 object to a local file asynchronously. It supports concurrent range requests for large objects and uses non-blocking disk writes.

```APIDOC
## `s3_client.download_file(Bucket, Key, Filename, ...)`

Downloads an S3 object to a local file asynchronously using concurrent range-get requests. Automatically splits large objects into parallel part downloads controlled by `Config.multipart_chunksize` and `Config.max_request_concurrency`. Uses `aiofiles` for non-blocking disk writes.

### Parameters

- `Bucket` (string): The name of the bucket containing the object.
- `Key` (string): The S3 object key (path) of the object to download.
- `Filename` (string): The local path where the file will be saved.
- `ExtraArgs` (dict, optional): A dictionary of additional arguments to pass to the S3 download operation (e.g., `VersionId`).
- `Config` (TransferConfig, optional): Configuration for the transfer, including `multipart_chunksize` and `max_request_concurrency`.

### Request Example

```python
import asyncio
import aioboto3
from boto3.s3.transfer import TransferConfig

async def main():
    session = aioboto3.Session()

    config = TransferConfig(
        multipart_chunksize=16 * 1024 * 1024,  # 16 MiB chunks
        max_request_concurrency=5,
    )

    async with session.client("s3", region_name="us-east-1") as s3:
        try:
            await s3.download_file(
                Bucket="my-data-bucket",
                Key="datasets/large_dataset.csv",
                Filename="/tmp/downloaded_dataset.csv",
                ExtraArgs={"VersionId": "abc123"},  # optional version pin
                Config=config,
            )
            print("Download complete")
        except Exception as e:
            print(f"Download failed: {e}")

asyncio.run(main())
```
```

--------------------------------

### Commit and push changes

Source: https://github.com/terricain/aioboto3/blob/main/CONTRIBUTING.rst

Stage, commit, and push your changes to your feature branch on GitHub.

```shell
git add .
```

```shell
git commit -m "Your detailed description of your changes."
```

```shell
git push origin name-of-your-bugfix-or-feature
```

--------------------------------

### Download S3 object to file object using aioboto3

Source: https://context7.com/terricain/aioboto3/llms.txt

Downloads an S3 object into a writable binary file-like object using concurrent range-get requests. Handles seekable and non-seekable streams, reordering chunks for the latter.

```python
import asyncio
import aioboto3
import aiofiles
from io import BytesIO

async def main():
    session = aioboto3.Session()

    async with session.client("s3", region_name="us-east-1") as s3:
        # Download into async file
        async with aiofiles.open("/tmp/output.bin", "wb") as f:
            await s3.download_fileobj("my-bucket", "path/to/object.bin", f)

        # Download into in-memory buffer
        buf = BytesIO()
        await s3.download_fileobj("my-bucket", "small/config.json", buf)
        buf.seek(0)
        import json
        config = json.load(buf)
        print(config)  # {"setting": "value"}

asyncio.run(main())
```

--------------------------------

### Upload File to S3 Asynchronously

Source: https://context7.com/terricain/aioboto3/llms.txt

Uploads a local file to S3 using `s3_client.upload_file`. It leverages `aiofiles` for non-blocking I/O and performs multipart uploads for large files. Progress can be tracked via a `Callback` function, and transfer configurations like `multipart_threshold` and `multipart_chunksize` can be customized.

```python
import asyncio
import aioboto3
from boto3.s3.transfer import TransferConfig

async def main():
    session = aioboto3.Session()

    uploaded_bytes = 0
    def progress_callback(bytes_so_far: int):
        nonlocal uploaded_bytes
        uploaded_bytes = bytes_so_far
        print(f"Uploaded: {bytes_so_far} bytes")

    config = TransferConfig(multipart_threshold=8 * 1024 * 1024, multipart_chunksize=8 * 1024 * 1024)

    async with session.client("s3", region_name="us-east-1") as s3:
        await s3.upload_file(
            Filename="/tmp/large_dataset.csv",
            Bucket="my-data-bucket",
            Key="datasets/large_dataset.csv",
            ExtraArgs={"ContentType": "text/csv", "ServerSideEncryption": "AES256"},
            Callback=progress_callback,
            Config=config,
        )
        print("Upload complete")

asyncio.run(main())
```

--------------------------------

### Download File from S3 Asynchronously

Source: https://context7.com/terricain/aioboto3/llms.txt

Downloads an S3 object to a local file using `s3_client.download_file`. This method supports concurrent range-get requests for efficient downloads of large objects, controlled by `multipart_chunksize` and `max_request_concurrency`. It uses `aiofiles` for non-blocking disk writes.

```python
import asyncio
import aioboto3
from boto3.s3.transfer import TransferConfig

async def main():
    session = aioboto3.Session()

    config = TransferConfig(
        multipart_chunksize=16 * 1024 * 1024,  # 16 MiB chunks
        max_request_concurrency=5,
    )

    async with session.client("s3", region_name="us-east-1") as s3:
        try:
            await s3.download_file(
                Bucket="my-data-bucket",
                Key="datasets/large_dataset.csv",
                Filename="/tmp/downloaded_dataset.csv",
                ExtraArgs={"VersionId": "abc123"},  # optional version pin
                Config=config,
            )
            print("Download complete")
        except Exception as e:
            print(f"Download failed: {e}")

asyncio.run(main())
```

--------------------------------

### Upload file object to S3 using aioboto3

Source: https://context7.com/terricain/aioboto3/llms.txt

Uploads a binary file-like object to S3. Supports regular and async file objects. Handles small payloads with put_object and large ones with multipart upload. An optional Processing callable can transform chunks.

```python
import asyncio
import aioboto3
import aiofiles

async def main():
    session = aioboto3.Session()

    async with session.client("s3", region_name="us-east-1") as s3:
        # Upload from async file
        async with aiofiles.open("/tmp/report.pdf", "rb") as f:
            await s3.upload_fileobj(
                f,
                "my-reports-bucket",
                "reports/2025/report.pdf",
                ExtraArgs={"ContentType": "application/pdf"},
            )

        # Upload from aiohttp response stream
        import aiohttp
        async with aiohttp.ClientSession() as http:
            async with http.get("https://example.com/data.bin") as resp:
                await s3.upload_fileobj(
                    resp.content,
                    "my-data-bucket",
                    "incoming/data.bin",
                )
        print("Both uploads complete")

asyncio.run(main())
```

--------------------------------

### s3_client.upload_file()

Source: https://context7.com/terricain/aioboto3/llms.txt

Uploads a local file to S3 asynchronously. It handles multipart uploads for large files and accepts extra arguments for S3 object configuration and a callback for progress tracking.

```APIDOC
## `s3_client.upload_file(Filename, Bucket, Key, ...)`

Uploads a local file to S3 asynchronously. Uses `aiofiles` for non-blocking file I/O and internally calls `upload_fileobj`. Performs a multipart upload if the file exceeds `Config.multipart_threshold` (default 8 MiB). Accepts `ExtraArgs` for metadata, ACLs, etc., and a `Callback` for progress tracking.

### Parameters

- `Filename` (string): The path to the file to upload.
- `Bucket` (string): The name of the bucket to upload to.
- `Key` (string): The S3 object key (path) in the bucket.
- `ExtraArgs` (dict, optional): A dictionary of additional arguments to pass to the S3 upload operation (e.g., `ContentType`, `ServerSideEncryption`).
- `Callback` (callable, optional): A function to be called with the number of bytes uploaded so far.
- `Config` (TransferConfig, optional): Configuration for the transfer, including `multipart_threshold` and `multipart_chunksize`.

### Request Example

```python
import asyncio
import aioboto3
from boto3.s3.transfer import TransferConfig

async def main():
    session = aioboto3.Session()

    uploaded_bytes = 0
    def progress_callback(bytes_so_far: int):
        nonlocal uploaded_bytes
        uploaded_bytes = bytes_so_far
        print(f"Uploaded: {bytes_so_far} bytes")

    config = TransferConfig(multipart_threshold=8 * 1024 * 1024, multipart_chunksize=8 * 1024 * 1024)

    async with session.client("s3", region_name="us-east-1") as s3:
        await s3.upload_file(
            Filename="/tmp/large_dataset.csv",
            Bucket="my-data-bucket",
            Key="datasets/large_dataset.csv",
            ExtraArgs={"ContentType": "text/csv", "ServerSideEncryption": "AES256"},
            Callback=progress_callback,
            Config=config,
        )
        print("Upload complete")

asyncio.run(main())
```
```

--------------------------------

### Concurrent AWS Requests with asyncio.gather

Source: https://context7.com/terricain/aioboto3/llms.txt

Shows how to perform multiple AWS API calls concurrently using `asyncio.gather`. This pattern is effective for reducing latency in I/O-bound operations.

```python
import asyncio
import aioboto3

async def main():
    session = aioboto3.Session()

    async with session.resource("dynamodb", region_name="us-east-1") as dynamo:
        table = await dynamo.Table("users")

        # Fan-out: fetch 50 users simultaneously
        user_ids = [f"user-{i:04d}" for i in range(50)]
        responses = await asyncio.gather(*[
            table.get_item(Key={"user_id": uid})
            for uid in user_ids
        ])
        users = [r["Item"] for r in responses if "Item" in r]
        print(f"Fetched {len(users)} users")

    # Fan-out S3 HEAD requests to check object existence
    async with session.client("s3", region_name="us-east-1") as s3:
        keys = [f"data/chunk_{i:04d}.parquet" for i in range(20)]
        results = await asyncio.gather(*[
            s3.head_object(Bucket="my-bucket", Key=k)
            for k in keys
        ], return_exceptions=True)
        existing = [k for k, r in zip(keys, results) if not isinstance(r, Exception)]
        print(f"Existing objects: {existing}")

asyncio.run(main())
```

--------------------------------

### S3 Resource - Bucket Object Collections

Source: https://context7.com/terricain/aioboto3/llms.txt

Manages S3 bucket objects using the S3 resource. Supports listing all objects, filtering by prefix, and deleting objects.

```APIDOC
## S3 Resource — Bucket Object Collections

The S3 `Bucket` resource exposes `objects.all()`, `objects.filter()`, and `objects.delete()` as async iterables and awaitables. Iterating `bucket.objects.all()` yields `ObjectSummary` instances with metadata available as coroutine properties.

### Methods

- **`bucket.objects.all()`**: Returns an async iterable of all objects in the bucket.
- **`bucket.objects.filter(Prefix='...')`**: Returns an async iterable of objects matching the specified prefix.
- **`bucket.objects.filter(Prefix='...').delete()`**: Deletes all objects matching the specified prefix.
- **`bucket.objects.all().delete()`**: Deletes all objects in the bucket.

### Request Example

```python
import asyncio
import aioboto3

async def main():
    session = aioboto3.Session()

    async with session.resource("s3", region_name="us-east-1") as s3:
        bucket = await s3.Bucket("my-bucket")

        async for obj in bucket.objects.all():
            print(obj.key, await obj.size)

        async for obj in bucket.objects.filter(Prefix="logs/2025/"):
            print(obj.key)

        await bucket.objects.filter(Prefix="temp/").delete()
        await bucket.objects.all().delete()

asyncio.run(main())
```
```

--------------------------------

### s3_client.download_fileobj

Source: https://context7.com/terricain/aioboto3/llms.txt

Downloads an S3 object into a writable binary file-like object using concurrent range-get requests. It supports both seekable and non-seekable streams, reordering chunks for non-seekable streams.

```APIDOC
## `s3_client.download_fileobj(Bucket, Key, Fileobj, ...)`

Downloads an S3 object into a writable binary file-like object using concurrent range-get requests. Handles both seekable (random-write) and non-seekable (sequential-write) streams. For non-seekable streams an internal queue reorders out-of-order chunks before writing.

### Parameters

- **Bucket**: The name of the bucket containing the object.
- **Key**: The key (path) of the object to download.
- **Fileobj**: The writable file-like object to download the object into.

### Request Example

```python
import asyncio
import aioboto3
import aiofiles
from io import BytesIO

async def main():
    session = aioboto3.Session()

    async with session.client("s3", region_name="us-east-1") as s3:
        async with aiofiles.open("/tmp/output.bin", "wb") as f:
            await s3.download_fileobj("my-bucket", "path/to/object.bin", f)

        buf = BytesIO()
        await s3.download_fileobj("my-bucket", "small/config.json", buf)
        buf.seek(0)
        import json
        config = json.load(buf)
        print(config)

asyncio.run(main())
```
```

--------------------------------

### Copy S3 object within or between buckets using aioboto3

Source: https://context7.com/terricain/aioboto3/llms.txt

Copies an S3 object using multipart upload for larger files or copy_object for smaller ones. A Semaphore limits concurrent part-copy requests. Supports cross-region copies with a SourceClient.

```python
import asyncio
import aioboto3
from boto3.s3.transfer import TransferConfig

async def main():
    session = aioboto3.Session()

    async with session.client("s3", region_name="us-east-1") as s3:
        # Copy within the same bucket
        await s3.copy(
            CopySource={"Bucket": "source-bucket", "Key": "originals/file.txt"},
            Bucket="dest-bucket",
            Key="copies/file.txt",
            ExtraArgs={"ServerSideEncryption": "AES256"},
        )

        # Cross-region copy using a second client as SourceClient
        async with session.client("s3", region_name="eu-west-1") as s3_eu:
            await s3.copy(
                CopySource={"Bucket": "eu-source-bucket", "Key": "data/archive.zip"},
                Bucket="us-dest-bucket",
                Key="archive/archive.zip",
                SourceClient=s3_eu,
                Config=TransferConfig(multipart_chunksize=32 * 1024 * 1024),
            )
        print("Copy complete")

asyncio.run(main())
```

--------------------------------

### Configure Retries for AWS Organizations Client

Source: https://github.com/terricain/aioboto3/blob/main/docs/usage.md

Illustrates how to configure custom retry behavior for an AWS Organizations client using AioConfig, increasing max attempts for concurrent requests.

```python3
import asyncio
from aioboto3 import Session
from aiobotocore.config import AioConfig

try_hard = AioConfig(retries={"max_attempts": 100})

async def main():
    coro = Session().client("organizations", config=try_hard)
    async with coro as client:
        resp_list = await asyncio.gather(
            *[client.list_roots() for _ in range(20)]
        )
        print([r["ResponseMetadata"]["RetryAttempts"] for r in resp_list])

asyncio.run(main())
```

--------------------------------

### Encrypting S3 Objects with KMS Managed Keys

Source: https://github.com/terricain/aioboto3/blob/main/docs/cse.md

This outlines the process for encrypting data before uploading to S3 using KMS Managed Keys. It involves generating a data key via KMS, encrypting the data, and assembling the necessary metadata for S3.

```text
Call the `generate_data_key` KMS API (with the encryption context) to get both an encrypted AES key and decypted AES key.
Generete IV’s. Encrypt your data. Assemble all the required metadata (use the KMS provided encrypted AES key for `x-amz-key-v2`), then push to S3.
```

--------------------------------

### s3_client.upload_fileobj

Source: https://context7.com/terricain/aioboto3/llms.txt

Uploads a binary file-like object to S3. It handles small payloads with put_object and large payloads with concurrent multipart uploads. An optional Processing callable can transform chunks before upload.

```APIDOC
## `s3_client.upload_fileobj(Fileobj, Bucket, Key, ...)`

Uploads a binary file-like object (sync or async) to S3. Supports both regular `BinaryIO` and async file objects (e.g., from `aiofiles`). Performs a `put_object` for small payloads or a concurrent multipart upload for large ones. Accepts an optional `Processing` callable to transform each chunk before upload (e.g., for compression).

### Parameters

- **Fileobj**: The file-like object to upload.
- **Bucket**: The name of the bucket to upload to.
- **Key**: The key (path) of the object in the bucket.
- **ExtraArgs**: Optional dictionary of additional arguments for the S3 operation (e.g., ContentType).

### Request Example

```python
import asyncio
import aioboto3
import aiofiles

async def main():
    session = aioboto3.Session()

    async with session.client("s3", region_name="us-east-1") as s3:
        async with aiofiles.open("/tmp/report.pdf", "rb") as f:
            await s3.upload_fileobj(
                f,
                "my-reports-bucket",
                "reports/2025/report.pdf",
                ExtraArgs={"ContentType": "application/pdf"},
            )

asyncio.run(main())
```
```