Server

class zonis.Server

Bases: RouteHandler

Parameters:

• using_fastapi_websockets (bool) – Defaults to False.

• override_key (Optional[str]) –

• secret_key (str) – Defaults to an emptry string.

async disconnect(identifier: str) → None

Disconnect a client connection.

Parameters:

identifier (str) – The client identifier to disconnect

Notes

This doesn’t yet tell the client to stop gracefully, this just removes it from our store.

async request(route: str, *, client_identifier: str = 'DEFAULT', **kwargs) → Any

Make a request to the provided IPC client.

Parameters:

• route (str) – The IPC route to call.

• client_identifier (Optional[str]) –

  The client to make a request to.

  This only applies in many to one setups or if you changed the default identifier.

• kwargs – All the arguments you wish to invoke the IPC route with.

Returns:

The data the IPC route returned.

Return type:

Any

Raises:

RequestFailed – The IPC request failed.

async request_all(route: str, **kwargs) → Dict[str, Any]

Issue a request to connected IPC clients.

Parameters:

• route (str) – The IPC route to call.

• kwargs – All the arguments you wish to invoke the IPC route with.

Returns:

A dictionary where the keys are the client identifiers and the values are the returned data.

The data could also be an instance of :py:class:RequestFailed:

Return type:

Dict[str, Any]

async parse_identify(packet: PacketT, websocket) → str

Parse a packet to establish a new valid client connection.

Parameters:

• packet (Packet) – The packet to read

• websocket – The websocket this connection is using

Returns:

The established clients identifier

Return type:

str

Raises:

• BaseZonisException – Unexpected WS issue

• DuplicateConnection – Duplicate connection without override keys

load_routes() → None

Loads all decorated routes.

register_class_instance_for_routes(instance, *routes) → None

Register a class instance for the given route.

When you turn a method on a class into an IPC route, you need to call this method with the instance of the class to use as well as the names of the routes for registration to work correctly.

Parameters:

• instance – The class instance the methods live on

• routes – A list of strings representing the names of the IPC routes for this class.

route(route_name: Optional[str] = None)

Turn an async function into a valid IPC route.

Parameters:

route_name (Optional[str]) – An optional name for this IPC route, defaults to the name of the function.

Raises:

DuplicateRoute – A route with this name already exists

Notes

If this is a method on a class, you will also need to use the register_class_instance_for_routes method for this to work as an IPC route.