File size: 1,290 Bytes
c12ef43 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | from __future__ import annotations
import typing
if typing.TYPE_CHECKING:
import wsproto
from .._models import Response
class HTTPXWSException(Exception):
"""
Base exception class for HTTPX WS.
"""
class WebSocketUpgradeError(HTTPXWSException):
"""
Raised when the initial connection didn't correctly upgrade to a WebSocket session.
"""
def __init__(self, response: Response) -> None:
self.response = response
class WebSocketDisconnect(HTTPXWSException):
"""
Raised when the server closed the WebSocket session.
Args:
code:
The integer close code to indicate why the connection has closed.
reason:
Additional reasoning for why the connection has closed.
"""
def __init__(self, code: int = 1000, reason: str | None = None) -> None:
self.code = code
self.reason = reason or ""
class WebSocketInvalidTypeReceived(HTTPXWSException):
"""
Raised when an event is not of the expected type.
"""
def __init__(self, event: wsproto.events.Event) -> None:
self.event = event
class WebSocketNetworkError(HTTPXWSException):
"""
Raised when a network error occured,
typically if the underlying stream has closed or timeout.
"""
|