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

Request Response Models

Name: tessl/pypi-http3
Author: tessl

Comprehensive objects for handling HTTP requests and responses with full data access, streaming support, content processing, and type safety. HTTP3 provides both synchronous and asynchronous versions of request and response models to support different programming patterns.

These models handle all aspects of HTTP communication including headers, cookies, content encoding/decoding, streaming, and provide convenient methods for common data access patterns.

Capabilities

Request Models

Synchronous Request

Represents an HTTP request with all necessary data and metadata.

class Request:
    def __init__(self, method, url, *, data=b"", params=None, headers=None):
        """
        Create an HTTP request.

        Parameters:
        - method (str): HTTP method (GET, POST, etc.)
        - url (URLTypes): Request URL
        - data (RequestData): Request body data
        - params (QueryParamTypes, optional): URL query parameters
        - headers (HeaderTypes, optional): HTTP headers
        """

    @property
    def method(self) -> str:
        """HTTP method (GET, POST, PUT, etc.)."""

    @property
    def url(self) -> URL:
        """Request URL object."""

    @property
    def headers(self) -> Headers:
        """Request headers."""

    @property
    def content(self) -> bytes:
        """Request body content as bytes."""

    def prepare(self):
        """Prepare the request for sending."""

Asynchronous Request

Asynchronous version of the Request class for non-blocking operations.

class AsyncRequest:
    def __init__(self, method, url, *, data=b"", params=None, headers=None):
        """
        Create an async HTTP request.

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

    @property
    def method(self) -> str:
        """HTTP method (GET, POST, PUT, etc.)."""

    @property
    def url(self) -> URL:
        """Request URL object."""

    @property
    def headers(self) -> Headers:
        """Request headers."""

    async def content(self) -> bytes:
        """Request body content as bytes (async)."""

    async def prepare(self):
        """Prepare the request for sending (async)."""

Response Models

Synchronous Response

Represents an HTTP response with comprehensive data access and processing capabilities.

class Response:
    @property
    def status_code(self) -> int:
        """HTTP status code (200, 404, etc.)."""

    @property
    def reason_phrase(self) -> str:
        """HTTP reason phrase (OK, Not Found, etc.)."""

    @property
    def protocol(self) -> str:
        """HTTP protocol version (HTTP/1.1, HTTP/2)."""

    @property
    def headers(self) -> Headers:
        """Response headers."""

    @property
    def cookies(self) -> Cookies:
        """Response cookies."""

    @property
    def url(self) -> URL:
        """Final URL (after redirects)."""

    @property
    def request(self) -> Request:
        """Original request object."""

    @property
    def content(self) -> bytes:
        """Response content as bytes."""

    @property
    def text(self) -> str:
        """Response content as decoded text."""

    @property
    def encoding(self) -> str:
        """Response content encoding."""

    @property
    def charset_encoding(self) -> str:
        """Character set encoding from headers."""

    def json(self) -> typing.Any:
        """
        Parse response content as JSON.
        
        Returns:
        Parsed JSON data
        
        Raises:
        json.JSONDecodeError: If content is not valid JSON
        """

    def read(self) -> bytes:
        """
        Read and return the response content.
        
        Returns:
        bytes: Complete response content
        """

    def stream(self) -> typing.Iterator[bytes]:
        """
        A byte-iterator over the decoded response content.
        This allows handling of gzip, deflate, and brotli encoded responses.
        
        Yields:
        bytes: Content chunks
        """

    def raise_for_status(self):
        """
        Raise an HTTPError for bad HTTP status codes.
        
        Raises:
        HTTPError: If status code indicates an error (4xx or 5xx)
        """

    @property
    def is_error(self) -> bool:
        """True if status code indicates an error (4xx or 5xx)."""

    @property
    def is_redirect(self) -> bool:
        """True if status code indicates a redirect (3xx)."""

    @property
    def is_client_error(self) -> bool:
        """True if status code indicates a client error (4xx)."""

    @property
    def is_server_error(self) -> bool:
        """True if status code indicates a server error (5xx)."""

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

Asynchronous Response

Asynchronous version of the Response class for non-blocking content processing.

class AsyncResponse:
    @property
    def status_code(self) -> int:
        """HTTP status code (200, 404, etc.)."""

    @property
    def reason_phrase(self) -> str:
        """HTTP reason phrase (OK, Not Found, etc.)."""

    @property
    def protocol(self) -> str:
        """HTTP protocol version (HTTP/1.1, HTTP/2)."""

    @property
    def headers(self) -> Headers:
        """Response headers."""

    @property
    def cookies(self) -> Cookies:
        """Response cookies."""

    @property
    def url(self) -> URL:
        """Final URL (after redirects)."""

    @property
    def request(self) -> AsyncRequest:
        """Original request object."""

    async def content(self) -> bytes:
        """Response content as bytes (async)."""

    async def text(self) -> str:
        """Response content as decoded text (async)."""

    @property
    def encoding(self) -> str:
        """Response content encoding."""

    @property
    def charset_encoding(self) -> str:
        """Character set encoding from headers."""

    async def json(self) -> typing.Any:
        """
        Parse response content as JSON (async).
        
        Returns:
        Parsed JSON data
        
        Raises:
        json.JSONDecodeError: If content is not valid JSON
        """

    async def stream(self) -> typing.AsyncIterator[bytes]:
        """
        A byte-iterator over the decoded response content (async).
        This allows handling of gzip, deflate, and brotli encoded responses.
        
        Yields:
        bytes: Content chunks
        """

    def raise_for_status(self):
        """
        Raise an HTTPError for bad HTTP status codes.
        
        Raises:
        HTTPError: If status code indicates an error (4xx or 5xx)
        """

    @property
    def is_error(self) -> bool:
        """True if status code indicates an error (4xx or 5xx)."""

    @property
    def is_redirect(self) -> bool:
        """True if status code indicates a redirect (3xx)."""

    @property
    def is_client_error(self) -> bool:
        """True if status code indicates a client error (4xx)."""

    @property
    def is_server_error(self) -> bool:
        """True if status code indicates a server error (5xx)."""

    async def close(self):
        """Close the response and release resources (async)."""

Content Types

Request Data Types

# Synchronous request data
RequestData = Union[dict, str, bytes, Iterator[bytes]]

# Asynchronous request data  
AsyncRequestData = Union[dict, str, bytes, AsyncIterator[bytes]]

# File upload types
RequestFiles = Dict[str, Union[
    IO[AnyStr],                           # Simple file object
    Tuple[str, IO[AnyStr]],              # (filename, file)
    Tuple[str, IO[AnyStr], str]          # (filename, file, content_type)
]]

Response Content Types

# Synchronous response content
ResponseContent = Union[bytes, Iterator[bytes]]

# Asynchronous response content
AsyncResponseContent = Union[bytes, AsyncIterator[bytes]]

Usage Examples

Basic Response Handling

import http3

# Synchronous response handling
response = http3.get('https://api.example.com/users')

print(f"Status: {response.status_code} {response.reason_phrase}")
print(f"Protocol: {response.protocol}")
print(f"Content-Type: {response.headers['content-type']}")

# Handle different content types
if response.headers.get('content-type', '').startswith('application/json'):
    data = response.json()
    print(f"JSON data: {data}")
else:
    text = response.text
    print(f"Text content: {text}")

Streaming Response Content

import http3

# Stream large responses
response = http3.get('https://example.com/large-file.zip', stream=True)

with open('downloaded-file.zip', 'wb') as f:
    for chunk in response.stream():
        f.write(chunk)

response.close()

Asynchronous Response Handling

import http3
import asyncio

async def fetch_data():
    async with http3.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
        
        # Async content access
        content = await response.content()
        text = await response.text()
        data = await response.json()
        
        # Async streaming
        async for chunk in response.stream():
            process_chunk(chunk)
        
        return data

asyncio.run(fetch_data())

Error Handling with Responses

import http3

try:
    response = http3.get('https://api.example.com/users')
    
    # Check for HTTP errors
    response.raise_for_status()
    
    # Or check status manually
    if response.is_error:
        print(f"HTTP Error: {response.status_code}")
    elif response.is_redirect:
        print(f"Redirected to: {response.url}")
    else:
        data = response.json()
        
except http3.ProtocolError as e:
    print(f"Protocol error: {e}")
except http3.DecodingError as e:
    print(f"Content decoding error: {e}")

Request Construction

import http3

# Manual request construction
request = http3.Request(
    'POST',
    'https://api.example.com/users',
    data=b'{"name": "John"}',
    headers={'Content-Type': 'application/json'}
)

# Send with client
with http3.Client() as client:
    response = client.send(request)

# Async request construction
async_request = http3.AsyncRequest(
    'GET',
    'https://api.example.com/users',
    params={'limit': 10}
)

Content Encoding and Decoding

import http3

response = http3.get('https://api.example.com/data')

# Automatic encoding detection
print(f"Detected encoding: {response.encoding}")
print(f"Charset from headers: {response.charset_encoding}")

# Force specific encoding
response.encoding = 'utf-8'
text = response.text

# Handle binary content
if response.headers.get('content-type', '').startswith('application/octet-stream'):
    binary_data = response.content
    # Process binary data

Install with Tessl CLI