tessl/pypi-aioboto3

Async boto3 wrapper providing asynchronous AWS SDK functionality

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Session Management

Name: tessl/pypi-aioboto3
Author: tessl

Core session functionality for creating and configuring AWS service clients and resources with async context management. The Session class serves as the main entry point for all aioboto3 operations.

Capabilities

Session Creation

Creates a new aioboto3 session with AWS credentials and configuration. The session manages authentication, region settings, and provides methods to create clients and resources.

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
    ):
        """
        Create a new aioboto3 session.
        
        Parameters:
        - aws_access_key_id: AWS access key ID
        - aws_secret_access_key: AWS secret access key
        - aws_session_token: AWS temporary session token
        - region_name: Default region when creating new connections
        - botocore_session: Use this aiobotocore session instead of creating new one
        - profile_name: The name of a profile to use from AWS credentials
        - aws_account_id: AWS account ID
        """

Resource Creation

Creates AWS service resources using async context managers. Resources provide high-level interfaces to AWS services with object-oriented access patterns.

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
):
    """
    Create a service resource.
    
    Parameters:
    - service_name: The name of a service (e.g. 's3', 'dynamodb', 'ec2')
    - region_name: The region to connect to
    - api_version: The API version to use
    - use_ssl: Whether to use SSL
    - verify: Whether to verify SSL certificates
    - endpoint_url: The complete URL to use for the constructed client
    - aws_access_key_id: The access key to use for this resource
    - aws_secret_access_key: The secret key to use for this resource
    - aws_session_token: The session token to use for this resource
    - config: Advanced configuration options
    
    Returns:
    ResourceCreatorContext: Async context manager that yields the service resource
    """

Client Creation

Creates AWS service clients using async context managers. Clients provide low-level access to AWS service APIs with direct method calls.

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
):
    """
    Create a low-level service client.
    
    Parameters are the same as resource() method.
    
    Returns:
    Async context manager that yields the service client
    """

Service Discovery

Methods for discovering available AWS services and resources supported by the session.

def get_available_services(self) -> list:
    """
    Get a list of service names that are available for creating clients.
    
    Returns:
    List of service names (strings)
    """

def get_available_resources(self) -> list:
    """
    Get a list of resource names that are available for creating resources.
    
    Returns:
    List of resource names (strings)
    """

Resource Context Management

The ResourceCreatorContext class manages the async lifecycle of AWS service resources.

class ResourceCreatorContext:
    async def __aenter__(self):
        """
        Async context manager entry point.
        
        Returns:
        The configured service resource instance
        """
    
    async def __aexit__(self, exc_type, exc, tb):
        """
        Async context manager exit point. Properly closes the underlying client.
        
        Parameters:
        - exc_type: Exception type if an exception occurred
        - exc: Exception value if an exception occurred
        - tb: Traceback if an exception occurred
        """

Usage Examples

Basic Session Usage

import aioboto3

# Create session with default credentials
session = aioboto3.Session()

# Create session with explicit credentials
session = aioboto3.Session(
    aws_access_key_id='your-access-key',
    aws_secret_access_key='your-secret-key',
    region_name='us-east-1'
)

# Create session with profile
session = aioboto3.Session(profile_name='development')

Using Resources

async def use_dynamodb():
    session = aioboto3.Session()
    
    async with session.resource('dynamodb', region_name='us-east-1') as dynamodb:
        table = await dynamodb.Table('my-table')
        # Use the table...

Using Clients

async def use_s3_client():
    session = aioboto3.Session()
    
    async with session.client('s3', region_name='us-east-1') as s3:
        response = await s3.list_buckets()
        print(response['Buckets'])

Service Discovery

session = aioboto3.Session()

# List available services for clients
services = session.get_available_services()
print("Available services:", services)

# List available resources
resources = session.get_available_resources()
print("Available resources:", resources)

Install with Tessl CLI