WebSocket API

WebSocket

WebSocket connection.

Constructor

WebSocket(scope: dict[str, Any], receive: Any, send: Any, manager: Any = None)

Properties

• path_params — Path parameters
• state — Connection state (CONNECTING, CONNECTED, CLOSING, CLOSED)
• accepted — Whether accepted
• closed — Whether closed

Methods

accept

async def accept(subprotocol: str | None = None, headers: list[tuple[bytes, bytes]] | None = None) -> None

receive

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

receive_text

async def receive_text() -> str

receive_bytes

async def receive_bytes() -> bytes

receive_json

async def receive_json() -> Any

send_text

async def send_text(value: str) -> None

send_bytes

async def send_bytes(value: bytes) -> None

send_json

async def send_json(value: Any) -> None

close

async def close(code: int = 1000, reason: str = "") -> None

join

async def join(room: str) -> None

leave

async def leave(room: str) -> None

broadcast_json

async def broadcast_json(room: str, value: Any) -> None

broadcast_text

async def broadcast_text(room: str, value: str) -> None

iter_json

def iter_json(self) -> AsyncIterator[Any]

iter_text

def iter_text(self) -> AsyncIterator[str]

WebSocketManager

WebSocket connection manager.

Methods

• async join(room, socket) — Add connection to room
• async leave(room, socket) — Remove connection from room
• async leave_all(socket) — Remove from all rooms
• async broadcast_text(room, message) — Broadcast text
• async broadcast_json(room, data) — Broadcast JSON
• async broadcast_bytes(room, data) — Broadcast bytes
• get_room_size(room) — Get number of connections
• get_rooms() — Get all room names
• get_connection_rooms(socket) — Get rooms a connection is in
• clear() — Clear all rooms and connections

WebSocketDisconnect

WebSocket disconnect exception.

WebSocketDisconnect(code: int = 1000, reason: str = "")

Attributes

• code — Close code
• reason — Close reason