Pure-Python HTTP/2 protocol implementation providing low-level connection and stream management
—
HTTP/2 settings negotiation, validation, and state management. Settings control connection behavior and are negotiated between peers during connection establishment and throughout the connection lifetime.
Standard HTTP/2 setting codes defined in the specification.
class SettingCodes(enum.IntEnum):
"""
All known HTTP/2 setting codes.
Values correspond to official HTTP/2 specification setting identifiers.
"""
HEADER_TABLE_SIZE = 0x1
"""Maximum size of header compression table used to decode header blocks."""
ENABLE_PUSH = 0x2
"""Enable or disable server push (0 to disable, 1 to enable)."""
MAX_CONCURRENT_STREAMS = 0x3
"""Maximum number of concurrent streams sender will allow."""
INITIAL_WINDOW_SIZE = 0x4
"""Sender's initial window size for stream-level flow control."""
MAX_FRAME_SIZE = 0x5
"""Size of largest frame payload sender is willing to receive."""
MAX_HEADER_LIST_SIZE = 0x6
"""Maximum size of header list sender is prepared to accept."""
ENABLE_CONNECT_PROTOCOL = 0x8
"""Enable connect protocol (0 to disable, 1 to enable)."""Object representing a setting value change.
class ChangedSetting:
"""
Represents a setting that has changed value.
Attributes:
setting: The setting code that changed
original_value: Previous value (None if first setting)
new_value: New value after change
"""
def __init__(
self,
setting: SettingCodes | int,
original_value: int | None,
new_value: int
):
"""
Initialize changed setting.
Args:
setting: Setting code (SettingCodes enum or int)
original_value: Previous value, None if not previously set
new_value: New value after change
"""
self.setting = setting
self.original_value = original_value
self.new_value = new_value
def __repr__(self) -> str:
"""String representation of the setting change."""Main settings management class implementing MutableMapping interface.
class Settings(MutableMapping[Union[SettingCodes, int], int]):
"""
HTTP/2 settings state management.
Manages local and remote settings with acknowledgment tracking,
validation, and default value handling. Settings changes are
tracked until acknowledged by the remote peer.
"""
def __init__(
self,
client: bool = True,
initial_values: dict[SettingCodes | int, int] | None = None
):
"""
Initialize settings object.
Args:
client: Whether this is for client-side (True) or server-side (False)
initial_values: Initial setting values to apply
"""
def acknowledge(self) -> dict[SettingCodes | int, ChangedSetting]:
"""
Acknowledge pending settings changes.
Called when SETTINGS ACK frame is received. Applies all pending
setting changes and returns what changed.
Returns:
Dictionary mapping setting codes to ChangedSetting objects
"""Direct access to individual HTTP/2 settings with validation.
@property
def header_table_size(self) -> int:
"""
HPACK dynamic table maximum size.
Range: 0 to 2^32-1
Default: 4096
"""
@header_table_size.setter
def header_table_size(self, value: int) -> None: ...
@property
def enable_push(self) -> int:
"""
Server push enablement.
Values: 0 (disabled) or 1 (enabled)
Default: 1 (enabled)
"""
@enable_push.setter
def enable_push(self, value: int) -> None: ...
@property
def initial_window_size(self) -> int:
"""
Initial flow control window size for new streams.
Range: 0 to 2^31-1
Default: 65535
"""
@initial_window_size.setter
def initial_window_size(self, value: int) -> None: ...
@property
def max_frame_size(self) -> int:
"""
Maximum frame payload size willing to receive.
Range: 16384 to 2^24-1
Default: 16384
"""
@max_frame_size.setter
def max_frame_size(self, value: int) -> None: ...
@property
def max_concurrent_streams(self) -> int:
"""
Maximum number of concurrent streams.
Range: 0 to 2^32-1
Default: unlimited (but h2 defaults to 100 for safety)
"""
@max_concurrent_streams.setter
def max_concurrent_streams(self, value: int) -> None: ...
@property
def max_header_list_size(self) -> int | None:
"""
Maximum header list size willing to accept.
Range: 0 to 2^32-1, or None if not set
Default: None (unlimited)
"""
@max_header_list_size.setter
def max_header_list_size(self, value: int | None) -> None: ...
@property
def enable_connect_protocol(self) -> int:
"""
Connect protocol enablement.
Values: 0 (disabled) or 1 (enabled)
Default: 0 (disabled)
"""
@enable_connect_protocol.setter
def enable_connect_protocol(self, value: int) -> None: ...Full dictionary-like interface for settings manipulation.
def __getitem__(self, key: SettingCodes | int) -> int:
"""Get setting value by code."""
def __setitem__(self, key: SettingCodes | int, value: int) -> None:
"""Set setting value by code."""
def __delitem__(self, key: SettingCodes | int) -> None:
"""Delete/reset setting to default value."""
def __iter__(self):
"""Iterate over setting codes."""
def __len__(self) -> int:
"""Number of configured settings."""
def __eq__(self, other) -> bool:
"""Compare settings objects for equality."""
def __ne__(self, other) -> bool:
"""Compare settings objects for inequality."""Helper functions for settings code conversion and validation.
def _setting_code_from_int(code: int) -> SettingCodes | int:
"""
Convert integer to SettingCodes enum if known, otherwise return int.
Args:
code: Integer setting code
Returns:
SettingCodes enum member if known, otherwise the original int
"""from h2.settings import Settings, SettingCodes
# Create settings for client
settings = Settings(client=True)
# Access individual settings
print(f"Initial window size: {settings.initial_window_size}")
print(f"Max frame size: {settings.max_frame_size}")
# Modify settings
settings.initial_window_size = 32768
settings.max_concurrent_streams = 50
# Use dictionary interface
settings[SettingCodes.ENABLE_PUSH] = 0 # Disable server pushfrom h2.settings import Settings, SettingCodes
# Initialize with custom defaults
initial_settings = {
SettingCodes.INITIAL_WINDOW_SIZE: 65536,
SettingCodes.MAX_CONCURRENT_STREAMS: 100,
SettingCodes.ENABLE_PUSH: 0
}
settings = Settings(client=True, initial_values=initial_settings)from h2.connection import H2Connection
from h2.settings import SettingCodes
conn = H2Connection()
# Update settings
new_settings = {
SettingCodes.INITIAL_WINDOW_SIZE: 32768,
SettingCodes.MAX_FRAME_SIZE: 32768
}
conn.update_settings(new_settings)
# Later, when SETTINGS ACK event is received
# (this happens automatically in H2Connection)
changed = conn.local_settings.acknowledge()
for setting_code, change in changed.items():
print(f"Setting {setting_code}: {change.original_value} -> {change.new_value}")from h2.settings import Settings, SettingCodes
from h2.exceptions import InvalidSettingsValueError
settings = Settings()
try:
# This will raise InvalidSettingsValueError
settings.max_frame_size = 1000 # Too small (minimum is 16384)
except InvalidSettingsValueError as e:
print(f"Invalid setting value: {e}")
print(f"Error code: {e.error_code}")
try:
# This will also raise InvalidSettingsValueError
settings.enable_push = 2 # Must be 0 or 1
except InvalidSettingsValueError as e:
print(f"Invalid setting value: {e}")from h2.settings import Settings
settings = Settings(client=True)
# Iterate over all settings
for setting_code in settings:
value = settings[setting_code]
print(f"{setting_code}: {value}")
# Check specific settings
if SettingCodes.MAX_HEADER_LIST_SIZE in settings:
print(f"Header list size limit: {settings.max_header_list_size}")
else:
print("No header list size limit set")
# Get all settings as dictionary
all_settings = dict(settings)
print(f"All settings: {all_settings}")Install with Tessl CLI
npx tessl i tessl/pypi-h2