tessl/pypi-hyperframe

Pure-Python HTTP/2 framing library providing comprehensive frame type support for creating, serializing, and parsing HTTP/2 frames.

—

Pending

Overview

Eval results

Files

Control Frames

Name: tessl/pypi-hyperframe
Author: tessl

HTTP/2 control frames for connection and stream management including SETTINGS, RST_STREAM, PRIORITY, PING, GOAWAY, and WINDOW_UPDATE frames. These frames manage the HTTP/2 connection lifecycle and provide flow control mechanisms.

Capabilities

SETTINGS Frames

SETTINGS frames convey configuration parameters that affect endpoint communication behavior. Settings are not negotiated and describe characteristics of the sending peer.

class SettingsFrame(Frame):
    def __init__(self, stream_id: int = 0, settings: dict[int, int] | None = None, **kwargs) -> None:
        """
        Create a SETTINGS frame.
        
        Parameters:
        - stream_id (int): Must be 0 (connection-level frame)
        - settings (dict[int, int] | None): Settings key-value pairs
        - flags (Iterable[str]): Frame flags ("ACK")
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is non-zero or ACK flag with non-empty settings
        """
    
    @property
    def settings(self) -> dict[int, int]:
        """Dictionary mapping setting identifiers to values."""
    
    # Settings constants
    HEADER_TABLE_SIZE: int  # 0x01
    ENABLE_PUSH: int  # 0x02
    MAX_CONCURRENT_STREAMS: int  # 0x03
    INITIAL_WINDOW_SIZE: int  # 0x04
    MAX_FRAME_SIZE: int  # 0x05
    MAX_HEADER_LIST_SIZE: int  # 0x06
    ENABLE_CONNECT_PROTOCOL: int  # 0x08
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags:

ACK (0x01): Acknowledges receipt of SETTINGS frame

RST_STREAM Frames

RST_STREAM frames allow abnormal termination of streams, indicating cancellation or error conditions.

class RstStreamFrame(Frame):
    def __init__(self, stream_id: int, error_code: int = 0, **kwargs) -> None:
        """
        Create a RST_STREAM frame.
        
        Parameters:
        - stream_id (int): Stream identifier (must be non-zero)
        - error_code (int): 32-bit error code for stream termination
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is zero
        """
    
    @property
    def error_code(self) -> int:
        """32-bit error code for stream reset."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags: None

PRIORITY Frames

PRIORITY frames specify sender-advised priority of streams and can be sent at any time to reprioritize existing streams.

class PriorityFrame(Priority, Frame):
    def __init__(self, stream_id: int, depends_on: int = 0, stream_weight: int = 0, 
                 exclusive: bool = False, **kwargs) -> None:
        """
        Create a PRIORITY frame.
        
        Parameters:
        - stream_id (int): Stream identifier (must be non-zero)
        - depends_on (int): Stream ID this stream depends on
        - stream_weight (int): Stream weight (0-255)
        - exclusive (bool): Whether this stream has exclusive dependency
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is zero
        """
    
    # Priority properties inherited from Priority mixin
    @property
    def depends_on(self) -> int: ...
    
    @property
    def stream_weight(self) -> int: ...
    
    @property
    def exclusive(self) -> bool: ...
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags: None

PING Frames

PING frames measure round-trip time and determine connection liveness. They can be sent from any endpoint.

class PingFrame(Frame):
    def __init__(self, stream_id: int = 0, opaque_data: bytes = b"", **kwargs) -> None:
        """
        Create a PING frame.
        
        Parameters:
        - stream_id (int): Must be 0 (connection-level frame)
        - opaque_data (bytes): Up to 8 bytes of opaque data
        - flags (Iterable[str]): Frame flags ("ACK")
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is non-zero
        - InvalidFrameError: If opaque_data exceeds 8 bytes
        """
    
    @property
    def opaque_data(self) -> bytes:
        """Opaque data payload (exactly 8 bytes when serialized)."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags:

ACK (0x01): Acknowledges receipt of PING frame

GOAWAY Frames

GOAWAY frames inform the remote peer to stop creating streams, initiating graceful connection shutdown.

class GoAwayFrame(Frame):
    def __init__(self, stream_id: int = 0, last_stream_id: int = 0, 
                 error_code: int = 0, additional_data: bytes = b"", **kwargs) -> None:
        """
        Create a GOAWAY frame.
        
        Parameters:
        - stream_id (int): Must be 0 (connection-level frame)
        - last_stream_id (int): Highest stream ID processed by sender
        - error_code (int): 32-bit error code for connection termination
        - additional_data (bytes): Additional diagnostic information
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is non-zero
        """
    
    @property
    def last_stream_id(self) -> int:
        """Highest stream ID that was processed by the sender."""
    
    @property
    def error_code(self) -> int:
        """32-bit error code for connection termination."""
    
    @property
    def additional_data(self) -> bytes:
        """Additional diagnostic data."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags: None

WINDOW_UPDATE Frames

WINDOW_UPDATE frames implement flow control at both stream and connection levels.

class WindowUpdateFrame(Frame):
    def __init__(self, stream_id: int, window_increment: int = 0, **kwargs) -> None:
        """
        Create a WINDOW_UPDATE frame.
        
        Parameters:
        - stream_id (int): Stream identifier (0 for connection-level)
        - window_increment (int): Flow control window increment (1 to 2^31-1)
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If window_increment is not in valid range
        """
    
    @property
    def window_increment(self) -> int:
        """Amount to increment the flow control window."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags: None

Usage Examples

SETTINGS Frame Operations

from hyperframe.frame import SettingsFrame

# Create SETTINGS frame with configuration
settings_frame = SettingsFrame(settings={
    SettingsFrame.HEADER_TABLE_SIZE: 4096,
    SettingsFrame.ENABLE_PUSH: 1,
    SettingsFrame.MAX_CONCURRENT_STREAMS: 100,
    SettingsFrame.INITIAL_WINDOW_SIZE: 65536,
    SettingsFrame.MAX_FRAME_SIZE: 16384,
    SettingsFrame.MAX_HEADER_LIST_SIZE: 8192
})

# Create SETTINGS ACK frame
settings_ack = SettingsFrame(flags=["ACK"])

print(f"Settings: {settings_frame.settings}")
print(f"ACK frame has no settings: {len(settings_ack.settings) == 0}")

# Access setting constants
print(f"Header table size setting ID: {SettingsFrame.HEADER_TABLE_SIZE}")

Stream Management Frames

from hyperframe.frame import RstStreamFrame, PriorityFrame

# Reset a stream due to error
rst_frame = RstStreamFrame(stream_id=1, error_code=8)  # CANCEL error

# Set stream priority
priority_frame = PriorityFrame(
    stream_id=3,
    depends_on=1,      # Depends on stream 1
    stream_weight=16,  # Weight of 16
    exclusive=False    # Non-exclusive dependency
)

print(f"RST_STREAM error code: {rst_frame.error_code}")
print(f"Priority: depends_on={priority_frame.depends_on}, weight={priority_frame.stream_weight}")

Connection Management

from hyperframe.frame import PingFrame, GoAwayFrame
import time

# Send PING to measure RTT
ping_data = int(time.time()).to_bytes(8, 'big')  # Timestamp as opaque data
ping_frame = PingFrame(opaque_data=ping_data)

# Respond to PING
ping_ack = PingFrame(opaque_data=ping_data, flags=["ACK"])

# Graceful connection shutdown
goaway_frame = GoAwayFrame(
    last_stream_id=5,
    error_code=0,  # NO_ERROR
    additional_data=b"Server shutdown"
)

print(f"PING data: {ping_frame.opaque_data}")
print(f"GOAWAY last stream: {goaway_frame.last_stream_id}")

Flow Control

from hyperframe.frame import WindowUpdateFrame

# Increase connection-level flow control window
connection_update = WindowUpdateFrame(
    stream_id=0,           # Connection level
    window_increment=32768  # Increment by 32KB
)

# Increase stream-level flow control window
stream_update = WindowUpdateFrame(
    stream_id=1,           # Specific stream
    window_increment=16384  # Increment by 16KB
)

print(f"Connection window increment: {connection_update.window_increment}")
print(f"Stream window increment: {stream_update.window_increment}")

Parsing Control Frames

from hyperframe.frame import Frame

# Parse SETTINGS frame
settings_bytes = (
    b'\\x00\\x00\\x0C\\x04\\x00\\x00\\x00\\x00\\x00'  # Header: 12 bytes, SETTINGS, no flags
    b'\\x00\\x01\\x00\\x00\\x10\\x00'                 # HEADER_TABLE_SIZE = 4096
    b'\\x00\\x02\\x00\\x00\\x00\\x01'                 # ENABLE_PUSH = 1
)
frame, length = Frame.parse_frame_header(settings_bytes[:9])
frame.parse_body(memoryview(settings_bytes[9:9 + length]))

print(f"Settings frame: {frame.settings}")

# Parse WINDOW_UPDATE frame
window_bytes = b'\\x00\\x00\\x04\\x08\\x00\\x00\\x00\\x00\\x01\\x00\\x00\\x40\\x00'
frame, length = Frame.parse_frame_header(window_bytes[:9])
frame.parse_body(memoryview(window_bytes[9:9 + length]))

print(f"Window update: stream={frame.stream_id}, increment={frame.window_increment}")

Error Handling

from hyperframe.frame import WindowUpdateFrame, SettingsFrame
from hyperframe.exceptions import InvalidDataError, InvalidFrameError

# Invalid window increment
try:
    WindowUpdateFrame(stream_id=1, window_increment=0)  # Must be >= 1
except InvalidDataError as e:
    print(f"Window update error: {e}")

# Invalid SETTINGS ACK with data
try:
    SettingsFrame(settings={1: 4096}, flags=["ACK"])  # ACK must have empty settings
except InvalidDataError as e:
    print(f"Settings error: {e}")

# Invalid PING data length
try:
    from hyperframe.frame import PingFrame
    PingFrame(opaque_data=b"too much data for ping frame")  # > 8 bytes
except InvalidFrameError as e:
    print(f"Ping error: {e}")

Install with Tessl CLI