tessl/pypi-aiohttp

Async http client/server framework (asyncio)

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

HTTP Client

Name: tessl/pypi-aiohttp
Author: tessl

Comprehensive HTTP client functionality built on asyncio for making HTTP requests with session management, connection pooling, and extensive configuration options. The client supports all HTTP methods, automatic redirects, cookie handling, authentication, and WebSocket connections.

Capabilities

Client Sessions

The primary interface for making HTTP requests. ClientSession provides connection pooling, cookie handling, and configuration management across multiple requests.

class ClientSession:
    def __init__(
        self,
        base_url=None,
        *,
        connector=None,
        loop=None,
        cookies=None,
        headers=None,
        proxy=None,
        proxy_auth=None,
        skip_auto_headers=None,
        auth=None,
        json_serialize=json.dumps,
        request_class=None,
        response_class=None,
        ws_response_class=None,
        version=None,
        cookie_jar=None,
        connector_owner=True,
        raise_for_status=False,
        read_timeout=None,
        conn_timeout=None,
        timeout=None,
        auto_decompress=True,
        trust_env=False,
        requote_redirect_url=True,
        trace_configs=None,
        read_bufsize=65536,
        max_line_size=8190,
        max_field_size=8190,
        fallback_charset_resolver=None
    ):
        """
        Create HTTP client session.
        
        Parameters:
        - base_url: Base URL for all requests
        - connector: Connection pool manager
        - loop: Event loop
        - cookies: Default cookies for all requests
        - headers: Default headers for all requests
        - proxy: Proxy URL
        - proxy_auth: Proxy authentication
        - skip_auto_headers: Headers to skip auto-generation
        - auth: Default authentication
        - json_serialize: JSON serialization function
        - request_class: Custom request class
        - response_class: Custom response class
        - ws_response_class: Custom WebSocket response class
        - version: HTTP version
        - cookie_jar: Cookie storage implementation
        - connector_owner: Whether session owns connector
        - raise_for_status: Auto-raise for HTTP errors
        - read_timeout: Read timeout (deprecated)
        - conn_timeout: Connection timeout (deprecated)
        - timeout: Timeout configuration
        - auto_decompress: Auto-decompress responses
        - trust_env: Trust environment variables for proxy
        - requote_redirect_url: Re-quote redirect URLs
        - trace_configs: Request tracing configurations
        - read_bufsize: Read buffer size
        - max_line_size: Maximum line size
        - max_field_size: Maximum header field size
        - fallback_charset_resolver: Charset resolver function
        """
    
    async def get(self, url, **kwargs):
        """Make GET request."""
    
    async def post(self, url, **kwargs):
        """Make POST request."""
    
    async def put(self, url, **kwargs):
        """Make PUT request."""
    
    async def patch(self, url, **kwargs):
        """Make PATCH request."""
    
    async def delete(self, url, **kwargs):
        """Make DELETE request."""
    
    async def head(self, url, **kwargs):
        """Make HEAD request."""
    
    async def options(self, url, **kwargs):
        """Make OPTIONS request."""
    
    async def request(self, method, url, **kwargs):
        """Make HTTP request with specified method."""
    
    async def ws_connect(self, url, **kwargs):
        """Establish WebSocket connection."""
    
    async def close(self):
        """Close session and cleanup resources."""
    
    def closed(self):
        """Check if session is closed."""

Standalone Request Function

Convenience function for making single HTTP requests without creating a session.

async def request(
    method,
    url,
    *,
    params=None,
    data=None,
    json=None,
    headers=None,
    cookies=None,
    auth=None,
    timeout=None,
    allow_redirects=True,
    proxy=None,
    ssl=None,
    **kwargs
):
    """
    Make HTTP request without creating session.
    
    Parameters:
    - method (str): HTTP method
    - url (str or URL): Request URL
    - params (dict): URL query parameters
    - data: Request body data
    - json: JSON request body
    - headers (dict): Request headers
    - cookies (dict): Request cookies
    - auth: Authentication credentials
    - timeout: Request timeout configuration
    - allow_redirects (bool): Follow redirects
    - proxy (str): Proxy URL
    - ssl: SSL context or verification settings
    
    Returns:
    ClientResponse: HTTP response object
    """

Request and Response Objects

Objects representing HTTP requests and responses with comprehensive data access methods.

class ClientRequest:
    def __init__(
        self,
        method,
        url,
        *,
        params=None,
        headers=None,
        data=None,
        cookies=None,
        auth=None,
        **kwargs
    ):
        """HTTP request representation."""
    
    @property
    def method(self):
        """Request HTTP method."""
    
    @property
    def url(self):
        """Request URL."""
    
    @property
    def headers(self):
        """Request headers."""

class ClientResponse:
    @property
    def status(self):
        """HTTP status code."""
    
    @property
    def reason(self):
        """HTTP reason phrase."""
    
    @property
    def headers(self):
        """Response headers."""
    
    @property
    def url(self):
        """Response URL."""
    
    @property
    def content_type(self):
        """Response content type."""
    
    async def read(self):
        """Read response body as bytes."""
    
    async def text(self, encoding=None):
        """Read response body as text."""
    
    async def json(self, **kwargs):
        """Parse response body as JSON."""
    
    def raise_for_status(self):
        """Raise exception for HTTP error status."""
    
    async def release(self):
        """Release response resources."""

Connection Management

Connection pooling and management with support for different transport types and SSL configuration.

class BaseConnector:
    def __init__(
        self,
        *,
        keepalive_timeout=30,
        enable_cleanup_closed=False,
        limit=100,
        limit_per_host=30,
        **kwargs
    ):
        """Base connector for connection management."""

class TCPConnector(BaseConnector):
    def __init__(
        self,
        *,
        ssl=None,
        use_dns_cache=True,
        ttl_dns_cache=10,
        family=0,
        local_addr=None,
        resolver=None,
        **kwargs
    ):
        """TCP connector for HTTP connections."""

class UnixConnector(BaseConnector):
    def __init__(self, path, **kwargs):
        """Unix domain socket connector."""

class NamedPipeConnector(BaseConnector):
    def __init__(self, path, **kwargs):
        """Named pipe connector (Windows)."""

Timeout Configuration

Comprehensive timeout management for different phases of HTTP requests.

class ClientTimeout:
    def __init__(
        self,
        total=None,
        sock_connect=None,
        sock_read=None
    ):
        """
        Configure request timeouts.
        
        Parameters:
        - total (float): Total request timeout
        - sock_connect (float): Socket connection timeout
        - sock_read (float): Socket read timeout
        """
    
    @property
    def total(self):
        """Total timeout value."""
    
    @property
    def sock_connect(self):
        """Socket connect timeout."""
    
    @property
    def sock_read(self):
        """Socket read timeout."""

Authentication

Authentication helpers for common HTTP authentication schemes.

class BasicAuth:
    def __init__(self, login, password, encoding='latin1'):
        """
        HTTP Basic authentication.
        
        Parameters:
        - login (str): Username
        - password (str): Password  
        - encoding (str): Text encoding
        """
    
    @property
    def login(self):
        """Authentication username."""
    
    @property
    def password(self):
        """Authentication password."""

class DigestAuthMiddleware:
    def __init__(self, username, password):
        """
        HTTP Digest authentication middleware.
        
        Parameters:
        - username (str): Username
        - password (str): Password
        """

Client Exceptions

Comprehensive exception hierarchy for HTTP client errors.

class ClientError(Exception):
    """Base exception for client errors."""

class ClientConnectionError(ClientError):
    """Connection-related errors."""

class ClientConnectorError(ClientConnectionError):
    """Connector-specific errors."""

class ClientConnectorDNSError(ClientConnectorError):
    """DNS resolution errors."""

class ClientConnectorSSLError(ClientConnectorError):
    """SSL/TLS connector errors."""

class ClientResponseError(ClientError):
    """HTTP response errors."""
    
    @property
    def status(self):
        """HTTP status code."""
    
    @property
    def message(self):
        """Error message."""

class ClientTimeout(ClientError):
    """Request timeout errors."""

class ContentTypeError(ClientError):
    """Content type parsing errors."""

class InvalidURL(ClientError):
    """Invalid URL errors."""

class TooManyRedirects(ClientError):
    """Excessive redirect errors."""

SSL and Security

SSL/TLS configuration and security features.

class Fingerprint:
    def __init__(self, fingerprint):
        """
        SSL certificate fingerprint verification.
        
        Parameters:
        - fingerprint (bytes): Expected certificate fingerprint
        """
    
    def check(self, transport):
        """Verify certificate fingerprint."""

Usage Examples

Basic HTTP Client

import asyncio
import aiohttp

async def fetch_data():
    async with aiohttp.ClientSession() as session:
        # GET request
        async with session.get('https://api.example.com/users') as response:
            users = await response.json()
        
        # POST request with JSON data
        user_data = {'name': 'John', 'email': 'john@example.com'}
        async with session.post('https://api.example.com/users', 
                               json=user_data) as response:
            result = await response.json()
        
        return users, result

users, result = asyncio.run(fetch_data())

Custom Session Configuration

import aiohttp
import asyncio

async def configured_client():
    # Custom timeout configuration
    timeout = aiohttp.ClientTimeout(total=30, sock_connect=10)
    
    # Custom headers for all requests
    headers = {'User-Agent': 'MyApp/1.0'}
    
    # Authentication
    auth = aiohttp.BasicAuth('username', 'password')
    
    async with aiohttp.ClientSession(
        timeout=timeout,
        headers=headers,
        auth=auth
    ) as session:
        async with session.get('https://api.example.com/protected') as response:
            return await response.text()

data = asyncio.run(configured_client())

Error Handling

import aiohttp
import asyncio

async def robust_request():
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get('https://api.example.com/data',
                                 timeout=aiohttp.ClientTimeout(total=10)) as response:
                response.raise_for_status()  # Raise for HTTP errors
                return await response.json()
    
    except aiohttp.ClientTimeout:
        print("Request timed out")
    except aiohttp.ClientConnectionError:
        print("Connection failed")
    except aiohttp.ClientResponseError as e:
        print(f"HTTP error: {e.status}")
    except Exception as e:
        print(f"Unexpected error: {e}")

result = asyncio.run(robust_request())

Install with Tessl CLI