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

Data and Header Frames

Name: tessl/pypi-hyperframe
Author: tessl

Implementation of HTTP/2 DATA and HEADERS frames for carrying application data and HTTP headers. These frames support padding, priority information, and proper stream association for efficient HTTP/2 communication.

Capabilities

DATA Frames

DATA frames carry arbitrary application data associated with a stream, such as HTTP request or response payloads.

class DataFrame(Padding, Frame):
    def __init__(self, stream_id: int, data: bytes = b"", **kwargs) -> None:
        """
        Create a DATA frame.
        
        Parameters:
        - stream_id (int): Stream identifier (must be non-zero)
        - data (bytes): Application data to carry
        - pad_length (int): Padding length if PADDED flag set
        - flags (Iterable[str]): Frame flags ("END_STREAM", "PADDED")
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is zero
        """
    
    @property
    def data(self) -> bytes:
        """Application data carried by this frame."""
    
    @property
    def flow_controlled_length(self) -> int:
        """
        Length of frame that counts toward flow control.
        
        Returns:
        int: Data length plus padding overhead
        """
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags:

END_STREAM (0x01): Indicates this is the last frame for the stream
PADDED (0x08): Indicates frame contains padding

HEADERS Frames

HEADERS frames carry HTTP header information and can open new streams. They support priority information and padding.

class HeadersFrame(Padding, Priority, Frame):
    def __init__(self, stream_id: int, data: bytes = b"", **kwargs) -> None:
        """
        Create a HEADERS frame.
        
        Parameters:
        - stream_id (int): Stream identifier (must be non-zero)
        - data (bytes): HPACK-encoded header block
        - pad_length (int): Padding length if PADDED flag set
        - depends_on (int): Stream dependency if PRIORITY flag set
        - stream_weight (int): Stream weight if PRIORITY flag set
        - exclusive (bool): Exclusive dependency if PRIORITY flag set
        - flags (Iterable[str]): Frame flags
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is zero
        """
    
    @property
    def data(self) -> bytes:
        """HPACK-encoded header block data."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags:

END_STREAM (0x01): Indicates this is the last frame for the stream
END_HEADERS (0x04): Indicates end of header list (no CONTINUATION frames follow)
PADDED (0x08): Indicates frame contains padding
PRIORITY (0x20): Indicates frame contains priority information

CONTINUATION Frames

CONTINUATION frames continue header block fragments when headers don't fit in a single HEADERS or PUSH_PROMISE frame.

class ContinuationFrame(Frame):
    def __init__(self, stream_id: int, data: bytes = b"", **kwargs) -> None:
        """
        Create a CONTINUATION frame.
        
        Parameters:
        - stream_id (int): Stream identifier (must be non-zero)
        - data (bytes): HPACK-encoded header block fragment
        - flags (Iterable[str]): Frame flags ("END_HEADERS")
        - **kwargs: Additional arguments for parent classes
        
        Raises:
        - InvalidDataError: If stream_id is zero
        """
    
    @property
    def data(self) -> bytes:
        """HPACK-encoded header block fragment."""
    
    # Inherited from Frame
    def serialize_body(self) -> bytes: ...
    def parse_body(self, data: memoryview) -> None: ...

Defined Flags:

END_HEADERS (0x04): Indicates end of header list

Usage Examples

Creating DATA Frames

from hyperframe.frame import DataFrame

# Simple DATA frame
data_frame = DataFrame(stream_id=1, data=b"Hello, World!")

# DATA frame with END_STREAM flag
final_frame = DataFrame(
    stream_id=1, 
    data=b"Final chunk", 
    flags=["END_STREAM"]
)

# DATA frame with padding
padded_frame = DataFrame(
    stream_id=1,
    data=b"Padded data",
    pad_length=10,
    flags=["PADDED"]
)

# Check flow control length
print(f"Flow control length: {padded_frame.flow_controlled_length}")

Creating HEADERS Frames

from hyperframe.frame import HeadersFrame

# Simple HEADERS frame
headers_frame = HeadersFrame(
    stream_id=1,
    data=b"\\x00\\x07:method\\x03GET",  # HPACK-encoded headers
    flags=["END_HEADERS"]
)

# HEADERS frame with priority
priority_headers = HeadersFrame(
    stream_id=1,
    data=b"\\x00\\x07:method\\x04POST",
    depends_on=0,
    stream_weight=16,
    exclusive=False,
    flags=["END_HEADERS", "PRIORITY"]
)

# HEADERS frame opening stream
stream_opener = HeadersFrame(
    stream_id=1,
    data=b"\\x00\\x07:method\\x03GET\\x00\\x05:path\\x01/",
    flags=["END_HEADERS", "END_STREAM"]  # GET request with no body
)

Working with CONTINUATION Frames

from hyperframe.frame import HeadersFrame, ContinuationFrame

# Large header block split across frames
header_data = b"\\x00\\x07:method\\x03GET" + b"\\x00" * 1000  # Large headers

# Initial HEADERS frame (without END_HEADERS)
headers_frame = HeadersFrame(
    stream_id=1,
    data=header_data[:500]
    # Note: no END_HEADERS flag
)

# CONTINUATION frame with remaining data
continuation_frame = ContinuationFrame(
    stream_id=1,
    data=header_data[500:],
    flags=["END_HEADERS"]  # End of header block
)

print(f"Headers frame: {len(headers_frame.data)} bytes")
print(f"Continuation frame: {len(continuation_frame.data)} bytes")

Parsing Data and Header Frames

from hyperframe.frame import Frame

# Parse DATA frame
data_bytes = b'\\x00\\x00\\x0D\\x00\\x01\\x00\\x00\\x00\\x01Hello, World!'
frame, length = Frame.parse_frame_header(data_bytes[:9])
frame.parse_body(memoryview(data_bytes[9:9 + length]))

print(f"Frame type: DATA")
print(f"Stream ID: {frame.stream_id}")
print(f"Data: {frame.data}")
print(f"END_STREAM: {'END_STREAM' in frame.flags}")

# Parse HEADERS frame with priority
headers_bytes = (
    b'\\x00\\x00\\x09\\x01\\x24\\x00\\x00\\x00\\x01'  # Header
    b'\\x00\\x00\\x00\\x00\\x10'                    # Priority (depends_on=0, weight=16)
    b'\\x00\\x07:method\\x03GET'                     # HPACK data
)
frame, length = Frame.parse_frame_header(headers_bytes[:9])
frame.parse_body(memoryview(headers_bytes[9:9 + length]))

print(f"Frame type: HEADERS")
print(f"Priority info: depends_on={frame.depends_on}, weight={frame.stream_weight}")
print(f"Exclusive: {frame.exclusive}")

Error Handling

from hyperframe.frame import DataFrame
from hyperframe.exceptions import InvalidDataError, InvalidPaddingError

# Invalid stream ID
try:
    DataFrame(stream_id=0, data=b"Invalid")  # DATA frames must have non-zero stream ID
except InvalidDataError as e:
    print(f"Error: {e}")

# Invalid padding
try:
    frame_data = b'\\x00\\x00\\x02\\x00\\x08\\x00\\x00\\x00\\x01\\xFF\\xFF'  # Padding > body length
    frame, length = Frame.parse_frame_header(frame_data[:9])
    frame.parse_body(memoryview(frame_data[9:9 + length]))
except InvalidPaddingError as e:
    print(f"Padding error: {e}")

Install with Tessl CLI