#src.fastware.websocket

#src.fastware.websocket

WebSocket helper class wrapping the raw ASGI scope/receive/send triple with typed accept, send, receive, and close methods for ergonomic usage.

#WebSocketDisconnect

Raised when a WebSocket client disconnects.

The code attribute carries the close code from the ASGI websocket.disconnect message (defaults to 1000 / normal closure).

#WebSocket

Wraps the raw ASGI (scope, receive, send) triple for WebSocket connections.

Handlers receive a WebSocket instance instead of the raw triple, providing convenient methods for accept/close/send/receive and properties for path_params, headers, and query_string.

#path_params

def path_params(self) -> dict[str, Any]

#headers

def headers(self) -> dict[str, str]

Parse ASGI headers into a case-preserving dict (first value wins).

#query_string

def query_string(self) -> str

#accept

async def accept(self, subprotocol: str | None=None) -> None

#close

async def close(self, code: int=1000) -> None

#send_json

async def send_json(self, data: Any) -> None

#send_bytes

async def send_bytes(self, data: bytes) -> None

#send_text

async def send_text(self, text: str) -> None

#_receive_data

async def _receive_data(self) -> dict[str, Any]

Receive a data message, raising WebSocketDisconnect on disconnect.

#receive_json

async def receive_json(self) -> Any

#receive_bytes

async def receive_bytes(self) -> bytes

#receive_text

async def receive_text(self) -> str

#receive_raw

async def receive_raw(self) -> dict[str, Any]

Return the raw ASGI message dict from the WebSocket connection.

The dict contains keys like "type", "bytes", "text" depending on the frame type. Useful for handlers that need to distinguish between binary and text frames without committing to one receive method.