tessl/pypi-http3

A next generation HTTP client for Python 3 with HTTP/2 support, async/await capabilities, and requests-compatible API.

—

Pending

Overview

Eval results

Files

Client Classes

Name: tessl/pypi-http3
Author: tessl

Persistent client instances that provide session-like behavior, connection pooling, and advanced configuration options. HTTP3 offers both synchronous and asynchronous client implementations, enabling efficient resource management and optimal performance for applications making multiple HTTP requests.

Clients maintain connection pools, handle cookies automatically, support authentication, and provide comprehensive configuration options for timeouts, SSL settings, and connection limits.

Capabilities

Synchronous Client

The primary synchronous HTTP client for applications requiring blocking I/O operations.

class Client:
    def __init__(self, auth=None, cookies=None, verify=True, cert=None, timeout=DEFAULT_TIMEOUT_CONFIG, pool_limits=DEFAULT_POOL_LIMITS, max_redirects=DEFAULT_MAX_REDIRECTS, base_url=None, dispatch=None, app=None, raise_app_exceptions=True, backend=None):
        """
        Initialize a synchronous HTTP client.

        Parameters:
        - auth (AuthTypes, optional): Default authentication for all requests
        - cookies (CookieTypes, optional): Default cookies for all requests
        - verify (VerifyTypes): SSL certificate verification (default: True)
        - cert (CertTypes, optional): Client SSL certificate
        - timeout (TimeoutTypes): Default timeout configuration
        - pool_limits (PoolLimits): Connection pool limits
        - max_redirects (int): Maximum number of redirects to follow
        - base_url (URLTypes, optional): Base URL for relative requests
        - dispatch (Dispatcher, optional): Custom request dispatcher
        - app (Callable, optional): WSGI/ASGI app for local testing
        - raise_app_exceptions (bool): Whether to raise app exceptions
        - backend (ConcurrencyBackend, optional): Concurrency backend
        """

    def request(self, method, url, *, data=None, files=None, json=None, params=None, headers=None, cookies=None, stream=False, auth=None, allow_redirects=True, cert=None, verify=None, timeout=None):
        """
        Send an HTTP request.

        Parameters: Same as module-level request functions
        Returns: Response object
        """

    def get(self, url, **kwargs):
        """Send a GET request. Returns Response object."""

    def post(self, url, **kwargs):
        """Send a POST request. Returns Response object."""

    def put(self, url, **kwargs):
        """Send a PUT request. Returns Response object."""

    def patch(self, url, **kwargs):
        """Send a PATCH request. Returns Response object."""

    def delete(self, url, **kwargs):
        """Send a DELETE request. Returns Response object."""

    def head(self, url, **kwargs):
        """Send a HEAD request. Returns Response object."""

    def options(self, url, **kwargs):
        """Send an OPTIONS request. Returns Response object."""

    def close(self):
        """Close the client and release all resources."""

    def __enter__(self):
        """Context manager entry."""

    def __exit__(self, exc_type, exc_value, traceback):
        """Context manager exit with automatic cleanup."""

Usage Example:

import http3

# Basic client usage
with http3.Client() as client:
    response = client.get('https://api.example.com/users')
    users = response.json()
    
    # Client maintains cookies and connections
    response = client.post('https://api.example.com/users', json={'name': 'John'})

# Client with authentication and custom configuration
client = http3.Client(
    auth=('username', 'password'),
    timeout=30.0,
    verify=False,  # Disable SSL verification
    base_url='https://api.example.com'
)

try:
    # Use relative URLs with base_url
    response = client.get('/users')
    response = client.post('/users', json={'name': 'Jane'})
finally:
    client.close()

# Advanced configuration
from http3 import PoolLimits, TimeoutConfig

client = http3.Client(
    pool_limits=PoolLimits(soft_limit=20, hard_limit=100),
    timeout=TimeoutConfig(connect_timeout=5.0, read_timeout=30.0),
    max_redirects=10
)

Asynchronous Client

The asynchronous HTTP client for non-blocking I/O operations and concurrent request handling.

class AsyncClient:
    def __init__(self, auth=None, cookies=None, verify=True, cert=None, timeout=DEFAULT_TIMEOUT_CONFIG, pool_limits=DEFAULT_POOL_LIMITS, max_redirects=DEFAULT_MAX_REDIRECTS, base_url=None, dispatch=None, app=None, raise_app_exceptions=True, backend=None):
        """
        Initialize an asynchronous HTTP client.

        Parameters: Same as Client.__init__()
        """

    async def request(self, method, url, *, data=None, files=None, json=None, params=None, headers=None, cookies=None, stream=False, auth=None, allow_redirects=True, cert=None, verify=None, timeout=None):
        """
        Send an HTTP request asynchronously.

        Parameters: Same as Client.request()
        Returns: AsyncResponse object
        """

    async def get(self, url, **kwargs):
        """Send a GET request asynchronously. Returns AsyncResponse object."""

    async def post(self, url, **kwargs):
        """Send a POST request asynchronously. Returns AsyncResponse object."""

    async def put(self, url, **kwargs):
        """Send a PUT request asynchronously. Returns AsyncResponse object."""

    async def patch(self, url, **kwargs):
        """Send a PATCH request asynchronously. Returns AsyncResponse object."""

    async def delete(self, url, **kwargs):
        """Send a DELETE request asynchronously. Returns AsyncResponse object."""

    async def head(self, url, **kwargs):
        """Send a HEAD request asynchronously. Returns AsyncResponse object."""

    async def options(self, url, **kwargs):
        """Send an OPTIONS request asynchronously. Returns AsyncResponse object."""

    async def close(self):
        """Close the client and release all resources asynchronously."""

    async def __aenter__(self):
        """Async context manager entry."""

    async def __aexit__(self, exc_type, exc_value, traceback):
        """Async context manager exit with automatic cleanup."""

Usage Example:

import http3
import asyncio

async def main():
    # Basic async client usage
    async with http3.AsyncClient() as client:
        response = await client.get('https://api.example.com/users')
        users = await response.json()
        
        # Make concurrent requests
        tasks = [
            client.get(f'https://api.example.com/users/{i}')
            for i in range(1, 11)
        ]
        responses = await asyncio.gather(*tasks)

    # Client with custom configuration
    client = http3.AsyncClient(
        auth=('username', 'password'),
        timeout=30.0,
        base_url='https://api.example.com'
    )
    
    try:
        response = await client.get('/users')
        data = await response.json()
    finally:
        await client.close()

asyncio.run(main())

Client Configuration Options

Authentication

# Basic authentication
client = http3.Client(auth=('username', 'password'))

# Custom authentication callable
def custom_auth(request):
    request.headers['Authorization'] = 'Bearer token'
    return request

client = http3.Client(auth=custom_auth)

Connection Pool Limits

from http3 import PoolLimits

# Custom pool limits
pool_limits = PoolLimits(
    soft_limit=10,    # Preferred number of connections per host
    hard_limit=50,    # Maximum connections per host
    pool_timeout=5.0  # Timeout waiting for connection from pool
)

client = http3.Client(pool_limits=pool_limits)

Timeout Configuration

from http3 import TimeoutConfig

# Granular timeout control
timeout = TimeoutConfig(
    connect_timeout=5.0,  # Connection establishment timeout
    read_timeout=30.0,    # Response reading timeout
    write_timeout=10.0    # Request sending timeout
)

client = http3.Client(timeout=timeout)

# Simple timeout (applies to all operations)
client = http3.Client(timeout=30.0)

SSL Configuration

# Disable SSL verification
client = http3.Client(verify=False)

# Custom CA bundle
client = http3.Client(verify='/path/to/ca-bundle.crt')

# Client certificate
client = http3.Client(cert='/path/to/client-cert.pem')

# Client certificate with private key
client = http3.Client(cert=('/path/to/cert.pem', '/path/to/key.pem'))

Base URL and Relative Requests

# Set base URL for relative requests
client = http3.Client(base_url='https://api.example.com/v1')

# All requests use the base URL
response = client.get('/users')  # Requests https://api.example.com/v1/users
response = client.post('/posts')  # Requests https://api.example.com/v1/posts

Cookie Persistence

# Initialize with cookies
client = http3.Client(cookies={'session': 'abc123'})

# Cookies are automatically maintained across requests
response = client.get('/login')  # Server sets cookies
response = client.get('/profile')  # Cookies sent automatically

WSGI/ASGI Integration

HTTP3 clients can make requests directly to WSGI or ASGI applications without network overhead:

from flask import Flask

app = Flask(__name__)

@app.route('/hello')
def hello():
    return {'message': 'Hello World'}

# Test Flask app directly
client = http3.Client(app=app)
response = client.get('/hello')
print(response.json())  # {'message': 'Hello World'}

Context Manager Usage

Both client types support context managers for automatic resource cleanup:

# Synchronous context manager
with http3.Client() as client:
    response = client.get('https://example.com')
    # Client automatically closed on exit

# Asynchronous context manager
async with http3.AsyncClient() as client:
    response = await client.get('https://example.com')
    # Client automatically closed on exit