# bocadillo.websockets

# WebSocket

WebSocket(self, scope:dict, receive:Callable[[], Awaitable[MutableMapping[str, Any]]], send:Callable[[MutableMapping[str, Any]], NoneType], auto_accept:bool=True, value_type:str=None, receive_type:str=None, send_type:str=None, caught_close_codes:Tuple[int, ...]=None)

Represents a WebSocket connection.

See also WebSockets.

Parameters

• scope (dict): ASGI scope.
• receive (callable): ASGI receive function.
• send (callable): ASGI send function.
• auto_accept (bool): whether to automatically accept the WebSocket connection request. Defaults to True. If False is passed, the connection must be accepted via .accept() or an async with block. Useful to perform conditionally .reject() a connection request or perform advanced error handling.
• value_type (str): The type of messages received or sent over the WebSocket. If given, overrides receive_type and send_type. Defaults to None.
• receive_type (str): The type of messages received over the WebSocket. Defaults to "text".
• send_type (str): The type of messages send over the WebSocket. Defaults to "text".
• caught_close_codes (tuple of int): Close codes of WebSocketDisconnect exceptions that should be caught and silenced. Defaults to (1000, 1001).

Attributes

• url (str-like): the URL which the WebSocket connection was made to.
• headers (dict): headers attached to the connection request.
• query_params (dict): parsed query parameters.

# accept

WebSocket.accept(self, subprotocol:str=None) -> None

Accept the connection request.

Parameters

• subprotocol (str): a specific WebSocket subprotocol that should be expected from clients. Defaults to None.

# reject

WebSocket.reject(self)

Reject the connection request.

This is equivalent to .close(403). Calling this before .accept() has undefined behavior.

# close

WebSocket.close(self, code:int=1000) -> None

Close the WebSocket connection.

Parameters

• code (int): a close code, defaults to 1000 (Normal Closure).

See Also

• CloseEvent specification

# ensure_closed

WebSocket.ensure_closed(self, code:int=1000)

Close the connection if it has not been closed already.

# receive_text

WebSocket.receive_text(self) -> str

Receive a message as text.

# send_text

WebSocket.send_text(self, data:str)

Send a message as text.

# receive_bytes

WebSocket.receive_bytes(self) -> bytes

Receive a message a bytes.

# send_bytes

WebSocket.send_bytes(self, data:bytes)

Send a message as bytes.

# receive_json

WebSocket.receive_json(self) -> Union[dict, list]

Receive a message as text and parse it to a JSON object.

Raises

• json.JSONDecodeError: if the received JSON is invalid.

# send_json

WebSocket.send_json(self, data:Union[dict, list])

Serialize an object to JSON and send it as text.

Raises

• TypeError: if the given data is not JSON serializable.

# receive_event

WebSocket.receive_event(self) -> MutableMapping[str, Any]

Receive a raw ASGI event.

# send_event

WebSocket.send_event(self, event:MutableMapping[str, Any])

Send a raw ASGI event.

# receive

WebSocket.receive(self) -> Union[str, bytes, list, dict]

Receive a message from the WebSocket.

Shortcut for receive_<self.receive_type>.

# send

WebSocket.send(self, message:Any)

Send a message over the WebSocket.

Shortcut for send_<self.send_type>.