or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client-side-encryption.mddynamodb.mdexperimental.mdindex.mds3-operations.mdsession-management.md
tile.json

tessl/pypi-aioboto3

Async boto3 wrapper providing asynchronous AWS SDK functionality

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/aioboto3@15.1.x

To install, run

npx @tessl/cli install tessl/pypi-aioboto3@15.1.0

index.mddocs/

aioboto3

An asynchronous wrapper for the AWS SDK for Python (boto3), enabling developers to use AWS services in async/await applications. aioboto3 combines the functionality of boto3 with aiobotocore to provide high-level resource APIs that work asynchronously, supporting services like DynamoDB, S3, Kinesis, SSM Parameter Store, and Athena with proper async context management.

Package Information

  • Package Name: aioboto3
  • Language: Python
  • Installation: pip install aioboto3

Optional Dependencies

  • S3 Client-Side Encryption: pip install aioboto3[s3cse]
  • Chalice Integration: pip install aioboto3[chalice]

Core Imports

import aioboto3

For session-based usage (recommended):

from aioboto3 import Session

Basic Usage

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

async def main():
    session = aioboto3.Session()
    
    # Using DynamoDB resource
    async with session.resource('dynamodb', region_name='us-east-1') as dynamo:
        table = await dynamo.Table('my-table')
        
        # Put an item
        await table.put_item(Item={'id': '123', 'data': 'hello'})
        
        # Query items
        response = await table.query(
            KeyConditionExpression=Key('id').eq('123')
        )
        print(response['Items'])
    
    # Using S3 client
    async with session.client('s3', region_name='us-east-1') as s3:
        # List buckets
        response = await s3.list_buckets()
        print(response['Buckets'])
        
        # Upload file
        await s3.upload_file('/path/to/file.txt', 'my-bucket', 'file.txt')

asyncio.run(main())

Architecture

aioboto3 extends the boto3 architecture with async capabilities:

  • Session: Main interface for creating async clients and resources, wraps aiobotocore session
  • ResourceCreatorContext: Async context manager that returns configured service resources
  • AIOBoto3ResourceFactory: Factory for creating async resource instances
  • Service Customizations: Specialized async implementations for DynamoDB, S3, and EC2

The library maintains compatibility with boto3's interface while adding async context managers for proper resource management and includes async implementations for operations that aren't natively async in boto3.

Capabilities

Session Management

Core session functionality for creating and configuring AWS service clients and resources with async context management. The Session class provides both low-level client access and high-level resource access to AWS services.

class Session:
    def __init__(
        self,
        aws_access_key_id: str = None,
        aws_secret_access_key: str = None,
        aws_session_token: str = None,
        region_name: str = None,
        botocore_session = None,
        profile_name: str = None,
        aws_account_id: str = None
    ): ...
    
    def client(
        self,
        service_name: str,
        region_name: str = None,
        api_version: str = None,
        use_ssl: bool = True,
        verify = None,
        endpoint_url: str = None,
        aws_access_key_id: str = None,
        aws_secret_access_key: str = None,
        aws_session_token: str = None,
        config = None
    ) -> AsyncContextManager: ...
    
    def resource(
        self,
        service_name: str,
        region_name: str = None,
        api_version: str = None,
        use_ssl: bool = True,
        verify = None,
        endpoint_url: str = None,
        aws_access_key_id: str = None,
        aws_secret_access_key: str = None,
        aws_session_token: str = None,
        config = None
    ) -> ResourceCreatorContext: ...

Session Management

DynamoDB Operations

Enhanced async DynamoDB operations including batch writing capabilities and async context management for table resources.

class CustomTableResource:
    def batch_writer(
        self,
        overwrite_by_pkeys = None,
        flush_amount: int = 25,
        on_exit_loop_sleep: int = 0
    ): ...

class BatchWriter:
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
    async def put_item(self, Item: dict, **kwargs): ...
    async def delete_item(self, Key: dict, **kwargs): ...

DynamoDB

S3 Operations

Async S3 operations including file transfers, object operations, and optional client-side encryption support.

async def upload_file(
    filename: str,
    bucket: str,
    key: str,
    callback = None,
    config = None
): ...

async def download_file(
    bucket: str,
    key: str,
    filename: str,
    callback = None,
    config = None
): ...

async def upload_fileobj(
    fileobj,
    bucket: str,
    key: str,
    callback = None,
    config = None
): ...

async def download_fileobj(
    bucket: str,
    key: str,
    fileobj,
    callback = None,
    config = None
): ...

S3 Operations

Client-Side Encryption

Advanced S3 client-side encryption capabilities for secure data storage with async support.

class S3CSEClient:
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
    async def put_object(self, **kwargs): ...
    async def get_object(self, **kwargs): ...

class S3CSEBucket:
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...

class S3CSEObject:
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...

Client-Side Encryption

Experimental Features

Experimental integrations including Chalice framework support for serverless applications.

class AsyncChalice:
    def __init__(
        self,
        *args,
        aioboto3_session: Session = None,
        **kwargs
    ): ...
    
    def __call__(self, event, context): ...

Experimental Features

Types

from typing import Optional, Callable, BinaryIO, Dict, Any, Union, AsyncContextManager
from abc import abstractmethod

TransferCallback = Callable[[int], None]

class _AsyncBinaryIO:
    @abstractmethod
    async def seek(self, offset: int, whence: int = 0) -> int: ...
    
    @abstractmethod
    async def write(self, s: Union[bytes, bytearray]) -> int: ...

AnyFileObject = Union[_AsyncBinaryIO, BinaryIO]

class ResourceCreatorContext:
    """Async context manager that creates and manages service resources."""
    def __init__(
        self,
        session,
        service_name: str,
        region_name: str,
        api_version: str,
        use_ssl: bool,
        verify,
        endpoint_url: str,
        aws_access_key_id: str,
        aws_secret_access_key: str,
        aws_session_token: str,
        config,
        resource_model
    ): ...
    
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc, tb): ...

class AIOBoto3ServiceResource:
    """Base class for all async service resources."""
    async def __aenter__(self) -> 'AIOBoto3ServiceResource': ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...