tessl/pypi-h2
Pure-Python HTTP/2 protocol implementation providing low-level connection and stream management
—
Pending
exceptions.mddocs/
Exception Handling
Comprehensive exception hierarchy for HTTP/2 protocol errors, stream management issues, and configuration problems. All exceptions provide detailed error information and appropriate HTTP/2 error codes.
Capabilities
Base Exception Classes
Foundation classes for the h2 exception hierarchy.
H2Error
Base class for all HTTP/2 related exceptions.
class H2Error(Exception):
"""Base class for all HTTP/2 exceptions."""ProtocolError
Base class for HTTP/2 protocol violations and specification non-compliance.
class ProtocolError(H2Error):
"""
HTTP/2 protocol violation detected.
Default error code: ErrorCodes.PROTOCOL_ERROR
"""
error_code = ErrorCodes.PROTOCOL_ERRORFrame Processing Exceptions
Exceptions related to HTTP/2 frame processing and validation.
FrameTooLargeError
Exception raised when a frame exceeds the maximum allowed size.
class FrameTooLargeError(ProtocolError):
"""
Frame exceeds maximum allowed size.
Default error code: ErrorCodes.FRAME_SIZE_ERROR
"""
error_code = ErrorCodes.FRAME_SIZE_ERRORFrameDataMissingError
Exception raised when a frame is missing required data fields.
class FrameDataMissingError(ProtocolError):
"""
Frame missing required data fields.
Default error code: ErrorCodes.FRAME_SIZE_ERROR
"""
error_code = ErrorCodes.FRAME_SIZE_ERRORUnsupportedFrameError
Exception raised when an unsupported frame is encountered in the current context.
class UnsupportedFrameError(ProtocolError):
"""Frame type not supported in current context."""Stream Management Exceptions
Exceptions related to HTTP/2 stream lifecycle and operations.
TooManyStreamsError
Exception raised when attempting to create more streams than allowed.
class TooManyStreamsError(ProtocolError):
"""Too many concurrent streams active."""StreamIDTooLowError
Exception raised when a stream ID violates ordering requirements.
class StreamIDTooLowError(ProtocolError):
"""
Stream ID is too low for the current connection state.
Attributes:
stream_id: The attempted stream ID
max_stream_id: Current highest allowed stream ID
"""
def __init__(self, stream_id: int, max_stream_id: int):
"""
Initialize with stream ID information.
Args:
stream_id: The attempted stream ID
max_stream_id: Current highest stream ID seen
"""
self.stream_id = stream_id
self.max_stream_id = max_stream_idNoAvailableStreamIDError
Exception raised when no stream IDs are available for new streams.
class NoAvailableStreamIDError(ProtocolError):
"""No stream IDs available for creating new streams."""NoSuchStreamError
Exception raised when attempting to operate on a non-existent stream.
class NoSuchStreamError(ProtocolError):
"""
Attempted operation on non-existent stream.
Attributes:
stream_id: The stream ID that doesn't exist
"""
def __init__(self, stream_id: int):
"""
Initialize with stream ID.
Args:
stream_id: The non-existent stream ID
"""
self.stream_id = stream_idStreamClosedError
Exception raised when attempting to operate on a closed stream.
class StreamClosedError(NoSuchStreamError):
"""
Attempted operation on closed stream.
Default error code: ErrorCodes.STREAM_CLOSED
Attributes:
stream_id: The closed stream ID
"""
error_code = ErrorCodes.STREAM_CLOSED
def __init__(self, stream_id: int):
"""
Initialize with stream ID.
Args:
stream_id: The closed stream ID
"""
self.stream_id = stream_id
self._events: list = []Flow Control Exceptions
Exceptions related to HTTP/2 flow control violations.
FlowControlError
Exception raised when flow control limits are violated.
class FlowControlError(ProtocolError):
"""
Flow control limits exceeded.
Default error code: ErrorCodes.FLOW_CONTROL_ERROR
"""
error_code = ErrorCodes.FLOW_CONTROL_ERRORSettings Exceptions
Exceptions related to HTTP/2 settings management and validation.
InvalidSettingsValueError
Exception raised when an invalid settings value is provided.
class InvalidSettingsValueError(ProtocolError, ValueError):
"""
Invalid value provided for HTTP/2 setting.
Attributes:
error_code: Specific error code for the invalid setting
"""
def __init__(self, msg: str, error_code: ErrorCodes):
"""
Initialize with error message and code.
Args:
msg: Error description
error_code: Specific HTTP/2 error code
"""
super().__init__(msg)
self.error_code = error_codeContent Validation Exceptions
Exceptions related to HTTP content and body validation.
InvalidBodyLengthError
Exception raised when actual content length doesn't match declared Content-Length.
class InvalidBodyLengthError(ProtocolError):
"""
Content-Length header doesn't match actual body length.
Attributes:
expected_length: Expected content length from header
actual_length: Actual content length received
"""
def __init__(self, expected: int, actual: int):
"""
Initialize with length information.
Args:
expected: Expected content length
actual: Actual content length received
"""
self.expected_length = expected
self.actual_length = actualRFC Compliance Exceptions
Exceptions for operations that are technically allowed but inadvisable.
RFC1122Error
Exception raised for operations that violate RFC 1122 "robustness principle".
class RFC1122Error(H2Error):
"""
Operation violates RFC 1122 robustness principle.
Used for operations that are technically allowed by HTTP/2 specification
but violate the principle of being conservative in what you send.
"""Security and DoS Protection Exceptions
Exceptions related to security and denial-of-service protection.
DenialOfServiceError
Exception raised when potential denial-of-service conditions are detected.
class DenialOfServiceError(ProtocolError):
"""
Potential denial-of-service condition detected.
Default error code: ErrorCodes.ENHANCE_YOUR_CALM
"""
error_code = ErrorCodes.ENHANCE_YOUR_CALMException Handling Patterns
Basic Exception Handling
from h2.connection import H2Connection
from h2.exceptions import ProtocolError, StreamClosedError, FlowControlError
conn = H2Connection()
try:
conn.send_headers(stream_id=1, headers=[...])
conn.send_data(stream_id=1, data=b"Hello")
except StreamClosedError as e:
print(f"Stream {e.stream_id} is closed")
except FlowControlError:
print("Flow control window exhausted")
# Wait for WindowUpdated event or implement backpressure
except ProtocolError as e:
print(f"Protocol error: {e}")
print(f"Error code: {e.error_code}")
# Consider closing connection
conn.close_connection(error_code=e.error_code)Stream Management Error Handling
from h2.exceptions import (
NoSuchStreamError, StreamIDTooLowError, NoAvailableStreamIDError
)
try:
stream_id = conn.get_next_available_stream_id()
conn.send_headers(stream_id=stream_id, headers=[...])
except NoAvailableStreamIDError:
print("No stream IDs available - connection exhausted")
# Need to open new connection
except StreamIDTooLowError as e:
print(f"Stream ID {e.stream_id} too low, max is {e.max_stream_id}")
# Use a higher stream ID
except NoSuchStreamError as e:
print(f"Stream {e.stream_id} doesn't exist")Settings and Configuration Error Handling
from h2.exceptions import InvalidSettingsValueError
from h2.errors import ErrorCodes
try:
conn.update_settings({
SettingCodes.INITIAL_WINDOW_SIZE: 65536,
SettingCodes.MAX_FRAME_SIZE: 32768
})
except InvalidSettingsValueError as e:
print(f"Invalid setting: {e}")
if e.error_code == ErrorCodes.FLOW_CONTROL_ERROR:
print("Flow control setting out of range")
elif e.error_code == ErrorCodes.FRAME_SIZE_ERROR:
print("Frame size setting invalid")Connection-Level Error Handling
from h2.exceptions import DenialOfServiceError, RFC1122Error
try:
# Process received data
events = conn.receive_data(network_data)
except DenialOfServiceError:
print("DoS protection triggered")
# Close connection gracefully
conn.close_connection(error_code=ErrorCodes.ENHANCE_YOUR_CALM)
except RFC1122Error as e:
print(f"RFC 1122 violation: {e}")
# Log but continue - remote peer being too liberal
except ProtocolError as e:
print(f"Fatal protocol error: {e}")
# Close connection immediately
conn.close_connection(error_code=e.error_code)Comprehensive Error Handler
from h2.exceptions import *
from h2.errors import ErrorCodes
import logging
logger = logging.getLogger(__name__)
def handle_h2_exception(e: Exception, conn: H2Connection) -> bool:
"""
Handle h2 exceptions with appropriate recovery actions.
Args:
e: The exception that occurred
conn: The H2Connection instance
Returns:
True if connection should continue, False if it should close
"""
if isinstance(e, StreamClosedError):
logger.warning(f"Operation on closed stream {e.stream_id}")
return True # Continue with other streams
elif isinstance(e, FlowControlError):
logger.warning("Flow control violation")
return True # Wait for window updates
elif isinstance(e, InvalidSettingsValueError):
logger.error(f"Invalid settings: {e}")
conn.close_connection(error_code=e.error_code)
return False
elif isinstance(e, DenialOfServiceError):
logger.critical("DoS protection triggered")
conn.close_connection(error_code=ErrorCodes.ENHANCE_YOUR_CALM)
return False
elif isinstance(e, ProtocolError):
logger.error(f"Protocol error: {e}")
conn.close_connection(error_code=getattr(e, 'error_code', ErrorCodes.PROTOCOL_ERROR))
return False
elif isinstance(e, H2Error):
logger.error(f"HTTP/2 error: {e}")
return True # Continue unless it's a protocol violation
else:
# Not an h2 exception
logger.error(f"Unexpected error: {e}")
return FalseInstall with Tessl CLI
npx tessl i tessl/pypi-h2