| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354 |
- from __future__ import annotations
- from enum import IntEnum
- from typing import Any
- class StreamErrorCode(IntEnum):
- """Error codes for HTTP/2 stream errors."""
- NO_ERROR = 0
- PROTOCOL_ERROR = 1
- INTERNAL_ERROR = 2
- FLOW_CONTROL_ERROR = 3
- SETTINGS_TIMEOUT = 4
- STREAM_CLOSED = 5
- FRAME_SIZE_ERROR = 6
- REFUSED_STREAM = 7
- CANCEL = 8
- COMPRESSION_ERROR = 9
- CONNECT_ERROR = 10
- ENHANCE_YOUR_CALM = 11
- INADEQUATE_SECURITY = 12
- HTTP_1_1_REQUIRED = 13
- @classmethod
- def _missing_(cls, value: object) -> Any: # noqa: ANN401, ARG003
- return cls.INTERNAL_ERROR
- class ConnectTimeout(ConnectionError, TimeoutError):
- """An error indicating a timeout while establishing a connection."""
- class RemoteProtocolError(Exception):
- """An error indicating the peer violated the HTTP protocol."""
- class StreamError(RemoteProtocolError):
- """An error representing an HTTP/2+ stream error."""
- code: StreamErrorCode
- """The error code associated with the stream error."""
- def __init__(self, message: str, code: StreamErrorCode | int) -> None:
- """Creates a new StreamError.
- Args:
- message: The error message.
- code: The stream error code.
- """
- super().__init__(message)
- if isinstance(code, int):
- code = StreamErrorCode(code)
- self.code = code
|