tessl/pypi-h11

A pure-Python, bring-your-own-I/O implementation of HTTP/1.1

—

Pending

Overview

Eval results

Files

HTTP Events

Name: tessl/pypi-h11
Author: tessl

Event classes representing different parts of HTTP messages including requests, responses, data, and connection lifecycle. All events inherit from the base Event class and are implemented as frozen dataclasses for immutability.

Capabilities

Base Event Class

Abstract base class for all h11 events.

class Event:
    """
    Base class for h11 events.
    
    Note:
        Do not subclass or instantiate directly. Use concrete event types.
    """

Request Events

Events representing HTTP requests from clients to servers.

class Request(Event):
    """
    The beginning of an HTTP request.
    
    Attributes:
        method (bytes): HTTP method (e.g., b"GET", b"POST")
        target (bytes): Request target (e.g., b"/index.html")
        headers (Headers): List of (name, value) header pairs
        http_version (bytes): HTTP version (e.g., b"1.1")
    """
    def __init__(
        self, 
        *, 
        method, 
        target, 
        headers, 
        http_version=b"1.1"
    ):
        """
        Create a new HTTP request event.
        
        Args:
            method (Union[bytes, str]): HTTP method
            target (Union[bytes, str]): Request target/path
            headers (Union[Headers, List[Tuple[bytes, bytes]], List[Tuple[str, str]]]): Request headers
            http_version (Union[bytes, str]): HTTP version (default: b"1.1")
        """

Usage Example:

# GET request
req = h11.Request(
    method=b'GET',
    target=b'/api/users',
    headers=[
        (b'host', b'api.example.com'),
        (b'user-agent', b'my-client/1.0'),
        (b'accept', b'application/json')
    ]
)

# POST request with body
req = h11.Request(
    method=b'POST',
    target=b'/api/users',
    headers=[
        (b'host', b'api.example.com'),
        (b'content-type', b'application/json'),
        (b'content-length', b'25')
    ]
)

Response Events

Events representing HTTP responses from servers to clients.

class InformationalResponse(Event):
    """
    HTTP informational response (status codes 100-199).
    
    Attributes:
        status_code (int): Status code (100-199)
        headers (Headers): Response headers
        http_version (bytes): HTTP version
        reason (bytes): Reason phrase
    """
    def __init__(
        self,
        *,
        status_code,
        headers=None,
        http_version=b"1.1",
        reason=b""
    ):
        """
        Create informational response event.
        
        Args:
            status_code (int): Status code (100-199)
            headers (Union[Headers, List[Tuple[bytes, bytes]], List[Tuple[str, str]]]): Headers (default: empty)
            http_version (Union[bytes, str]): HTTP version (default: b"1.1")
            reason (Union[bytes, str]): Reason phrase (default: auto-generated)
        """

class Response(Event):
    """
    Beginning of HTTP response (status codes 200-999).
    
    Attributes:
        status_code (int): Status code (200-999)
        headers (Headers): Response headers
        http_version (bytes): HTTP version
        reason (bytes): Reason phrase
    """
    def __init__(
        self,
        *,
        status_code,
        headers=None,
        http_version=b"1.1",
        reason=b""
    ):
        """
        Create response event.
        
        Args:
            status_code (int): Status code (200-999)
            headers (Union[Headers, List[Tuple[bytes, bytes]], List[Tuple[str, str]]]): Headers (default: empty)
            http_version (Union[bytes, str]): HTTP version (default: b"1.1")
            reason (Union[bytes, str]): Reason phrase (default: auto-generated)
        """

Usage Examples:

# 100 Continue response
info_resp = h11.InformationalResponse(status_code=100)

# 200 OK response
resp = h11.Response(
    status_code=200,
    headers=[
        (b'content-type', b'application/json'),
        (b'content-length', b'42'),
        (b'cache-control', b'no-cache')
    ]
)

# 404 Not Found response
resp = h11.Response(
    status_code=404,
    headers=[(b'content-type', b'text/plain')],
    reason=b'Not Found'
)

Message Body Events

Events representing HTTP message body data and message completion.

class Data(Event):
    """
    Part of an HTTP message body.
    
    Attributes:
        data (bytes): Message body data chunk
        chunk_start (bool): Start of chunked encoding chunk
        chunk_end (bool): End of chunked encoding chunk
    """
    def __init__(self, *, data, chunk_start=False, chunk_end=False):
        """
        Create data event.
        
        Args:
            data (Union[bytes, bytearray, memoryview]): Body data
            chunk_start (bool): Mark start of chunk (default: False)
            chunk_end (bool): Mark end of chunk (default: False)
        """

class EndOfMessage(Event):
    """
    End of an HTTP message.
    
    Attributes:
        headers (Headers): Trailing headers (default: empty)
    """
    def __init__(self, *, headers=None):
        """
        Create end-of-message event.
        
        Args:
            headers (Union[Headers, List[Tuple[bytes, bytes]], List[Tuple[str, str]]]): 
                Trailing headers (default: empty list)
        """

Usage Examples:

# Send body data
data_event = h11.Data(data=b'{"user": "john", "age": 30}')

# Send chunked data
chunk_start = h11.Data(data=b'Hello', chunk_start=True)
chunk_data = h11.Data(data=b' World')  
chunk_end = h11.Data(data=b'!', chunk_end=True)

# End message (no trailing headers)
eom = h11.EndOfMessage()

# End message with trailing headers
eom = h11.EndOfMessage(headers=[
    (b'x-checksum', b'abc123'),
    (b'x-timestamp', b'1609459200')
])

Connection Lifecycle Events

Events representing connection state changes.

class ConnectionClosed(Event):
    """
    Indicates the sender has closed their outgoing connection.
    
    Note:
        This event has no attributes and represents a clean connection closure.
        Receiving this event means no more data will be sent by the peer.
    """
    def __init__(self):
        """Create connection closed event."""

Usage Example:

# Indicate connection is closing
closed_event = h11.ConnectionClosed()

# Handle in event loop
event = conn.next_event()
if isinstance(event, h11.ConnectionClosed):
    print("Peer closed connection")
    # Clean up resources

Event Processing Patterns

Client Request Pattern

import h11

conn = h11.Connection(h11.CLIENT)

# 1. Send request headers
request = h11.Request(
    method=b'POST',
    target=b'/upload',  
    headers=[
        (b'host', b'example.com'),
        (b'content-type', b'application/json'),
        (b'transfer-encoding', b'chunked')
    ]
)
data = conn.send(request)
send_to_socket(data)

# 2. Send body data
body_chunk = h11.Data(data=b'{"key": "value"}')
data = conn.send(body_chunk)
send_to_socket(data)

# 3. End message
eom = h11.EndOfMessage()
data = conn.send(eom)
send_to_socket(data)

Server Response Pattern

import h11

conn = h11.Connection(h11.SERVER)

# 1. Send response headers
response = h11.Response(
    status_code=200,
    headers=[
        (b'content-type', b'text/html'),
        (b'content-length', b'13')
    ]
)
data = conn.send(response)
send_to_socket(data)

# 2. Send body
body = h11.Data(data=b'Hello, World!')
data = conn.send(body)
send_to_socket(data)

# 3. End message
eom = h11.EndOfMessage()
data = conn.send(eom)
send_to_socket(data)

Event Processing Loop

while True:
    # Get more data from socket
    raw_data = receive_from_socket()
    if not raw_data:
        break
        
    # Feed to connection
    conn.receive_data(raw_data)
    
    # Process all available events
    while True:
        event = conn.next_event()
        
        if event is h11.NEED_DATA:
            break  # Need more data from socket
        elif event is h11.PAUSED:
            conn.start_next_cycle()
            break
        elif isinstance(event, h11.Request):
            print(f"Request: {event.method} {event.target}")
        elif isinstance(event, h11.Response):
            print(f"Response: {event.status_code}")
        elif isinstance(event, h11.Data):
            print(f"Body data: {len(event.data)} bytes")
        elif isinstance(event, h11.EndOfMessage):
            print("Message complete")
        elif isinstance(event, h11.ConnectionClosed):
            print("Connection closed")
            return

Install with Tessl CLI