API

WebSocket Endpoint

The WebSocket Endpoint

class socketsundso.endpoints.HandlingEndpointMeta(*args: str, **kwargs: Any)

Bases: type

Metaclass for WebSocketHandlingEndpoint

All this does is put every Handler inside of WebSocketHandlingEndpoint into WebSocketHandlingEndpoint.handlers

class socketsundso.endpoints.WebSocketHandlingEndpoint(scope: MutableMapping[str, Any], receive: Callable[[], Awaitable[MutableMapping[str, Any]]], send: Callable[[MutableMapping[str, Any]], Awaitable[None]])

Bases: object

The WebSocketHandlingEndpoint is a class for the creation of a simple JSON-based WebSocket API

This class is based on starlette.endpoints.WebSocketEndpoint Incoming messages have to be based on WebSocketEventMessage

dispatch() will call handlers based on the incoming WebSocketEventMessage.type. If the handler returns something it will be send to the client.

To register a method as handler decorate it with socketsundso.handler.on_event() or on_event().

You can override on_connect() and on_disconnect() to change what happens when clients connect or disconnect.

async dispatch() → None

Handles the lifecycle of a WebSocket connection and calls on_connect(), the on_receive() and on_disconnect() repectively.

Note

This method will be called by starlette. You shouldn’t need to think about it.

If the client sends a JSON payload that can’t be validated by WebSocketEventMessage or on_receive() raises an ValidationError or json.decoder.JSONDecodeError the errors will be send to the client via send_exception().

handlers: Dict[str, socketsundso.handler.Handler] = {}

async on_connect() → None

Override to handle an incoming websocket connection

Note

Don’t forget to call self.websocket.accept() to accept the connection.

async on_disconnect(close_code: int) → None

Override to handle a disconnecting websocket

classmethod on_event(*args: Any, **kwargs: Any) → Callable

See also

Same arguments as socketsundso.handler.on_event().

async on_receive(message: socketsundso.models.WebSocketEventMessage) → None

Called by dispatch() whenever a message arrives.

Calls the Handler.handle() for the event and calls send_json() with the response.

async send_exception(exc: Exception) → None

Formats the exc and sends it to the client (via send_json())

Override if you don’t wnat to send any Exceptions to the client or want to format them differently.

async send_json(response: Any) → None

Calls fastapi.encoders.jsonable_encoder() and passes result to starlette.websockets.WebSocket.send_json().

Override to handle outgoing messages differently. For example you could handle handler response differently based on their type.

websocket

WebSocket instance

Handler

Handler and related methods

class socketsundso.handler.Handler(event: str, method: Callable, response_model: Optional[Type[pydantic.main.BaseModel]] = None)

Bases: object

Class representation of a handler. It holds information about the handler, e.g. model (based on pydantic.BaseModel), event, etc

Parameters

• event (str) – name of the event this handler is for

• method (Callable) – method this handler will call

• response_model (pydantic.BaseModel) – handle() will parse the return value of method into this model. If no model is given a default response model will be created.

async handle(msg: socketsundso.models.WebSocketEventMessage) → pydantic.main.BaseModel | None

Will be called by WebSocketHandlingEndpoint.on_receive()

Parameters

msg (WebSocketEventMessage) – will be validated against model

Returns

response_model

Return type

WebSocketEventMessage

model

Based on WebSocketEventMessage with fields for the parameters of method. Will be used for input validation.

socketsundso.handler.on_event(event: Callable) → Callable

socketsundso.handler.on_event(event: str | None = None, response_model: Optional[Type[pydantic.main.BaseModel]] = None) → Callable

Turns a typing.Callable into a Handler Decorator to be used in subclasses of WebSocketHandlingEndpoint Declares a method as handler for event

Takes the same parameters as Handler

Technical Note: Since it’s impossible to get the class of an unbound function this decorator just sets some attributes on the function. The registration as handler happens in HandlingEndpointMeta__new__()

Models

pydantic models used in this project

class socketsundso.models.WebSocketEventMessage(*, type: str, **extra_data: Any)

Bases: pydantic.main.BaseModel

BaseModel of an incoming WebSocketEvent

All other arguments of a handler will be added dynamically.

Note

When validating incoming messages, type will be replaced with a typing.Literal for all registered event types. So basically this whole class is kind of pointless.

class Config

Bases: object

extra = 'allow'

use_enum_values = True

type: str