tessl/pypi-botocore

Low-level, data-driven core of boto 3 providing foundational AWS service access.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Client API

Name: tessl/pypi-botocore
Author: tessl

Service clients provide direct access to AWS APIs with dynamically generated methods based on service models. Clients handle request construction, response parsing, and service-specific logic for all AWS services.

Capabilities

Base Client Class

Foundation class for all AWS service clients with common functionality.

class BaseClient:
    def __init__(
        self,
        serializer,
        endpoint,
        response_parser,
        event_emitter,
        request_signer,
        service_model: ServiceModel,
        loader,
        client_config: Config,
        partition,
        exceptions_factory,
        endpoint_ruleset_resolver=None,
        user_agent_creator=None
    ):
        """
        Initialize AWS service client. 
        
        Note: BaseClient is typically created via session.create_client() rather than direct instantiation.
        
        Args:
            serializer: Request serialization handler
            endpoint: Service endpoint handler
            response_parser: Response parsing handler  
            event_emitter: Event emission system
            request_signer: Request signing handler
            service_model: Service model defining API operations
            loader: Data loader for service models
            client_config: Client configuration object
            partition: AWS partition information
            exceptions_factory: Exception creation factory
            endpoint_ruleset_resolver: Endpoint rule resolution
            user_agent_creator: User agent string creation
        """

Client Metadata

Access client configuration and metadata through the meta property.

class BaseClient:
    @property
    def meta(self) -> ClientMeta:
        """Client metadata and configuration access."""

class ClientMeta:
    @property
    def service_model(self) -> ServiceModel:
        """Service model defining API operations."""
    
    @property
    def region_name(self) -> str:
        """AWS region name for this client."""
    
    @property
    def client_config(self) -> Config:
        """Client configuration object."""
    
    @property
    def config(self) -> dict:
        """Resolved configuration dictionary."""
    
    @property
    def events(self) -> HierarchicalEmitter:
        """Event emitter for client events."""

URL Generation

Generate presigned URLs for direct access to AWS resources.

class BaseClient:
    def generate_presigned_url(
        self,
        ClientMethod: str,
        Params: dict = None,
        ExpiresIn: int = 3600,
        HttpMethod: str = None
    ) -> str:
        """
        Generate presigned URL for client method.
        
        Args:
            ClientMethod: Client method name (e.g., 'get_object')
            Params: Parameters for the operation
            ExpiresIn: URL expiration time in seconds
            HttpMethod: HTTP method for the URL
            
        Returns:
            str: Presigned URL
        """
    
    def generate_presigned_post(
        self,
        Bucket: str,
        Key: str,
        Fields: dict = None,
        Conditions: List = None,
        ExpiresIn: int = 3600
    ) -> dict:
        """
        Generate presigned POST for S3 uploads.
        
        Args:
            Bucket: S3 bucket name
            Key: S3 object key
            Fields: Form fields to include
            Conditions: Policy conditions
            ExpiresIn: Expiration time in seconds
            
        Returns:
            dict: Presigned POST data with 'url' and 'fields'
        """

Pagination Support

Access and use pagination for operations that return large result sets.

class BaseClient:
    def can_paginate(self, operation_name: str) -> bool:
        """
        Check if operation supports pagination.
        
        Args:
            operation_name: AWS operation name
            
        Returns:
            bool: True if operation supports pagination
        """
    
    def get_paginator(self, operation_name: str) -> Paginator:
        """
        Get paginator for operation.
        
        Args:
            operation_name: AWS operation name
            
        Returns:
            Paginator: Paginator instance for operation
        """

Waiter Support

Access waiters for polling resource states until desired conditions are met.

class BaseClient:
    def get_waiter(self, waiter_name: str) -> Waiter:
        """
        Get waiter for resource state polling.
        
        Args:
            waiter_name: Waiter name
            
        Returns:
            Waiter: Waiter instance
        """

Connection Management

Manage client HTTP connections and cleanup.

class BaseClient:
    def close(self) -> None:
        """Close client HTTP connections and cleanup resources."""

Client Factory

Factory class for creating service clients.

class ClientCreator:
    def create_client(
        self,
        service_name: str,
        region_name: str,
        api_version: str = None,
        use_ssl: bool = True,
        verify: Union[bool, str] = None,
        endpoint_url: str = None,
        credentials: Credentials = None,
        scoped_config: dict = None,
        client_config: Config = None,
        api_version_override: str = None
    ) -> BaseClient:
        """
        Create AWS service client.
        
        Args:
            service_name: AWS service name
            region_name: AWS region name
            api_version: API version to use
            use_ssl: Use SSL/TLS for requests
            verify: SSL certificate verification
            endpoint_url: Custom endpoint URL
            credentials: AWS credentials
            scoped_config: Configuration dictionary
            client_config: Client configuration
            api_version_override: Override API version
            
        Returns:
            BaseClient: Configured service client
        """

Dynamic Service Methods

Service clients dynamically generate methods based on service models. Each AWS service operation becomes a client method with the following pattern:

Operation Method Pattern

def operation_name(self, **kwargs) -> dict:
    """
    AWS service operation.
    
    Args:
        **kwargs: Operation-specific parameters as defined in service model
        
    Returns:
        dict: Operation response data
        
    Raises:
        ClientError: Service-specific errors
        BotoCoreError: Client-level errors
    """

Common Operation Categories

List Operations: Return collections of resources

# Example: S3 list_buckets
response = s3_client.list_buckets()
buckets = response['Buckets']

# Example: EC2 describe_instances  
response = ec2_client.describe_instances()
reservations = response['Reservations']

Get Operations: Retrieve specific resources

# Example: S3 get_object
response = s3_client.get_object(Bucket='mybucket', Key='mykey')
body = response['Body'].read()

# Example: EC2 describe_instance_attribute
response = ec2_client.describe_instance_attribute(
    InstanceId='i-1234567890abcdef0',
    Attribute='instanceType'
)

Create Operations: Create new resources

# Example: S3 create_bucket
response = s3_client.create_bucket(
    Bucket='mynewbucket',
    CreateBucketConfiguration={'LocationConstraint': 'us-west-2'}
)

# Example: EC2 run_instances
response = ec2_client.run_instances(
    ImageId='ami-12345678',
    MinCount=1,
    MaxCount=1,
    InstanceType='t2.micro'
)

Update Operations: Modify existing resources

# Example: S3 put_object
response = s3_client.put_object(
    Bucket='mybucket',
    Key='mykey',
    Body=b'Hello World'
)

# Example: EC2 modify_instance_attribute
ec2_client.modify_instance_attribute(
    InstanceId='i-1234567890abcdef0',
    InstanceType={'Value': 't2.small'}
)

Delete Operations: Remove resources

# Example: S3 delete_object
s3_client.delete_object(Bucket='mybucket', Key='mykey')

# Example: EC2 terminate_instances
response = ec2_client.terminate_instances(
    InstanceIds=['i-1234567890abcdef0']
)

Usage Examples

Basic Client Usage

from botocore.session import get_session

# Create session and client
session = get_session()
s3_client = session.create_client('s3', region_name='us-east-1')

# Use client methods
response = s3_client.list_buckets()
for bucket in response['Buckets']:
    print(bucket['Name'])

Client Configuration

from botocore.config import Config

# Create client with custom configuration
config = Config(
    connect_timeout=30,
    read_timeout=30,
    retries={'max_attempts': 3}
)

client = session.create_client(
    's3',
    region_name='us-west-2',
    config=config
)

Presigned URL Generation

# Generate presigned URL for S3 object
url = s3_client.generate_presigned_url(
    'get_object',
    Params={'Bucket': 'mybucket', 'Key': 'myfile.txt'},
    ExpiresIn=3600  # 1 hour
)

# Generate presigned POST for file uploads
presigned_post = s3_client.generate_presigned_post(
    Bucket='mybucket',
    Key='uploads/${filename}',
    ExpiresIn=3600
)

Error Handling

from botocore.exceptions import ClientError, NoCredentialsError

try:
    response = s3_client.head_bucket(Bucket='nonexistent-bucket')
except ClientError as e:
    error_code = e.response['Error']['Code']
    if error_code == 'NoSuchBucket':
        print("Bucket does not exist")
    elif error_code == 'Forbidden':
        print("Access denied")
except NoCredentialsError:
    print("AWS credentials not found")

Connection Management

# Close client connections when done
try:
    # Use client for operations
    response = s3_client.list_buckets()
    # ... more operations
finally:
    s3_client.close()

Install with Tessl CLI