_errors.py 1.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354
  1. from __future__ import annotations
  2. from enum import IntEnum
  3. from typing import Any
  4. class StreamErrorCode(IntEnum):
  5. """Error codes for HTTP/2 stream errors."""
  6. NO_ERROR = 0
  7. PROTOCOL_ERROR = 1
  8. INTERNAL_ERROR = 2
  9. FLOW_CONTROL_ERROR = 3
  10. SETTINGS_TIMEOUT = 4
  11. STREAM_CLOSED = 5
  12. FRAME_SIZE_ERROR = 6
  13. REFUSED_STREAM = 7
  14. CANCEL = 8
  15. COMPRESSION_ERROR = 9
  16. CONNECT_ERROR = 10
  17. ENHANCE_YOUR_CALM = 11
  18. INADEQUATE_SECURITY = 12
  19. HTTP_1_1_REQUIRED = 13
  20. @classmethod
  21. def _missing_(cls, value: object) -> Any: # noqa: ANN401, ARG003
  22. return cls.INTERNAL_ERROR
  23. class ConnectTimeout(ConnectionError, TimeoutError):
  24. """An error indicating a timeout while establishing a connection."""
  25. class RemoteProtocolError(Exception):
  26. """An error indicating the peer violated the HTTP protocol."""
  27. class StreamError(RemoteProtocolError):
  28. """An error representing an HTTP/2+ stream error."""
  29. code: StreamErrorCode
  30. """The error code associated with the stream error."""
  31. def __init__(self, message: str, code: StreamErrorCode | int) -> None:
  32. """Creates a new StreamError.
  33. Args:
  34. message: The error message.
  35. code: The stream error code.
  36. """
  37. super().__init__(message)
  38. if isinstance(code, int):
  39. code = StreamErrorCode(code)
  40. self.code = code