WebSocket

WebSocket connection handling.

WebSocket connection handling for ASGI applications.

Classes:

WebSocketState — IntEnum: CONNECTING(0), CONNECTED(1), DISCONNECTED(2) WebSocket — wraps ASGI receive/send with Pythonic API

Flow: WebSocket(scope, receive, send) → accept() → send/receive → close()

Strict typing: receive_text() rejects bytes, receive_bytes() rejects text. Optional TYTX support: receive_typed()/send_typed() for typed serialization.

class genro_asgi.websocket.WebSocket(scope, receive, send)

Bases: object

WebSocket connection wrapper for ASGI.

Provides a Pythonic interface for handling WebSocket connections through the ASGI protocol. Manages connection lifecycle (accept, close) and message sending/receiving.

scope

The ASGI scope dictionary.

connection_state

Current WebSocketState.

Example

>>> async def handler(scope, receive, send):
...     ws = WebSocket(scope, receive, send)
...     await ws.accept()
...     message = await ws.receive_text()
...     await ws.send_text(f"Echo: {message}")
...     await ws.close()

__init__(scope, receive, send)

Initialize WebSocket connection wrapper.

Parameters:

• scope (MutableMapping[str, Any]) – ASGI WebSocket scope dictionary.

• receive (Callable[[], Awaitable[MutableMapping[str, Any]]]) – ASGI receive callable.

• send (Callable[[MutableMapping[str, Any]], Awaitable[None]]) – ASGI send callable.

Raises:

ValueError – If scope type is not “websocket”.

async accept(subprotocol=None, headers=None)

Accept the WebSocket connection.

Must be called before sending or receiving messages. Consumes the websocket.connect message and sends websocket.accept.

Parameters:

• subprotocol (str | None) – Optional subprotocol to negotiate.

• headers (Mapping[str, str] | None) – Optional headers to include in accept response.

Raises:

RuntimeError – If not in CONNECTING state or unexpected message.

Return type:

None

property accepted_subprotocol: str | None

The subprotocol selected in accept(), or None.

property asgi_receive: Callable[[], Awaitable[MutableMapping[str, Any]]]

The ASGI receive callable.

property asgi_send: Callable[[MutableMapping[str, Any]], Awaitable[None]]

The ASGI send callable.

property client: Address | None

Client address (host, port) if available.

async close(code=1000, reason='')

Close the WebSocket connection.

Idempotent: safe to call multiple times.

Parameters:

• code (int) – WebSocket close code (default 1000).

• reason (str) – Optional close reason.

Raises:

RuntimeError – If connection not accepted yet.

Return type:

None

property connection_state: WebSocketState

Current connection state.

property headers: Headers

Connection headers (case-insensitive).

async iter_bytes()

Async iterator yielding binary messages.

Yields:

Each binary message until disconnect.

Raises:

TypeError – If text message received.

async iter_text()

Async iterator yielding text messages.

Yields:

Each text message until disconnect.

Raises:

TypeError – If binary message received.

property path: str

URL path component.

property query_params: QueryParams

Query string parameters.

async receive_bytes()

Receive a binary message.

Return type:

bytes

Returns:

The binary message content.

Raises:

• RuntimeError – If not connected.

• TypeError – If received text data.

• WebSocketDisconnect – If client disconnected.

async receive_json()

Receive and parse a JSON text message.

Return type:

Any

Returns:

The parsed JSON value.

Raises:

• RuntimeError – If not connected.

• TypeError – If received binary data.

• json.JSONDecodeError – If not valid JSON.

• WebSocketDisconnect – If client disconnected.

async receive_text()

Receive a text message.

Return type:

str

Returns:

The text message content.

Raises:

• RuntimeError – If not connected.

• TypeError – If received binary data.

• WebSocketDisconnect – If client disconnected.

async receive_typed()

Receive JSON with optional TYTX hydration.

Return type:

dict[str, Any]

Returns:

Parsed dict, hydrated if TYTX marker present.

Raises:

• ImportError – If genro-tytx not installed.

• RuntimeError – If not connected.

property scheme: str

URL scheme (‘ws’ or ‘wss’).

property scope: MutableMapping[str, Any]

The raw ASGI scope dictionary.

async send_bytes(data)

Send a binary message.

Parameters:

data (bytes) – Bytes to send.

Raises:

RuntimeError – If not connected.

Return type:

None

async send_json(data)

Serialize data to JSON and send.

Parameters:

data (Any) – JSON-serializable data.

Raises:

• RuntimeError – If not connected.

• TypeError – If data not serializable.

Return type:

None

async send_text(data)

Send a text message.

Parameters:

data (str) – Text to send.

Raises:

RuntimeError – If not connected.

Return type:

None

async send_typed(data)

Serialize with TYTX and send with marker.

Parameters:

data (dict[str, Any]) – Dict with potentially typed values.

Raises:

• ImportError – If genro-tytx not installed.

• RuntimeError – If not connected.

Return type:

None

property state: State

Per-connection state container.

property subprotocols: tuple[str, ...]

Subprotocols requested by the client (immutable).

property url: URL

Full WebSocket URL.

Constructed lazily from scope fields: scheme, server, root_path, path, and query_string.

Returns:

URL object representing the complete WebSocket URL.

class genro_asgi.websocket.WebSocketState(*values)

Bases: IntEnum

WebSocket connection state.

Represents the three possible states of a WebSocket connection:

• CONNECTING: Initial state, before accept() is called

• CONNECTED: After accept(), can send and receive messages

• DISCONNECTED: After close() or client disconnect

Example

>>> ws = WebSocket(scope, receive, send)
>>> ws.connection_state == WebSocketState.CONNECTING
True
>>> await ws.accept()
>>> ws.connection_state == WebSocketState.CONNECTED
True

CONNECTED = 1

CONNECTING = 0

DISCONNECTED = 2