websocket#

class starlite.connection.websocket.WebSocket#

Bases: Generic[User, Auth], ASGIConnection[WebsocketRouteHandler, User, Auth]

The Starlite WebSocket class.

__init__(scope: Scope, receive: Receive = <function empty_receive>, send: Send = <function empty_send>) → None#

Initialize WebSocket.

Parameters:

  • scope – The ASGI connection scope.

  • receive – The ASGI receive function.

  • send – The ASGI send function.

receive_wrapper(receive: Receive) → Receive#

Wrap receive to set ‘self.connection_state’ and validate events.

Parameters:

receive – The ASGI receive function.

Returns:

An ASGI receive function.

send_wrapper(send: Send) → Send#

Wrap send to ensure that state is not disconnected.

Parameters:

send – The ASGI send function.

Returns:

An ASGI send function.

async accept(subprotocols: Optional[str] = None, headers: Optional[Union[Headers, Dict[str, Any], List[Tuple[bytes, bytes]]]] = None) → None#

Accept the incoming connection. This method should be called before receiving data.

Parameters:

  • subprotocols – Websocket sub-protocol to use.

  • headers – Headers to set on the data sent.

Returns:

None

async close(code: int = 1000, reason: Optional[str] = None) → None#

Send an ‘websocket.close’ event.

Parameters:

  • code – Status code.

  • reason – Reason for closing the connection

Returns:

None

async receive_data(mode: Literal['text']) → str#
async receive_data(mode: Literal['binary']) → bytes

Receive an ‘websocket.receive’ event and returns the data stored on it.

Parameters:

mode – The respective event key to use.

Returns:

The event’s data.

async receive_text() → str#

Receive data as text.

Returns:

A string.

async receive_bytes() → bytes#

Receive data as bytes.

Returns:

A byte-string.

async receive_json(mode: Literal['text', 'binary'] = 'text') → Any#

Receive data and loads it into JSON using orson.

Parameters:

mode – Either text or binary.

Returns:

An arbitrary value

async send_data(data: Union[str, bytes], mode: Literal['text', 'binary'] = 'text', encoding: str = 'utf-8') → None#

Send a ‘websocket.send’ event.

Parameters:

  • data – Data to send.

  • mode – The respective event key to use.

  • encoding – Encoding to use when converting bytes / str.

Returns:

None

async send_text(data: bytes, encoding: str = 'utf-8') → None#
async send_text(data: str) → None

Send data using the text key of the send event.

Parameters:

  • data – Data to send

  • encoding – Encoding to use for binary data.

Returns:

None

async send_bytes(data: bytes) → None#
async send_bytes(data: str, encoding: str = 'utf-8') → None

Send data using the bytes key of the send event.

Parameters:

  • data – Data to send

  • encoding – Encoding to use for binary data.

Returns:

None

async send_json(data: ~typing.Any, mode: ~typing.Literal['text', 'binary'] = 'text', encoding: str = 'utf-8', serializer: Serializer = <function default_serializer>) → None#

Send data as JSON.

Parameters:

  • data – A value to serialize.

  • mode – Either text or binary.

  • encoding – Encoding to use for binary data.

  • serializer – A serializer function.

Returns:

None