Router#

zonis.route(route_name: Optional[str] = None)[source]#

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

class zonis.Router[source]#

Bases: object

A router class for enabling two-way communication down a single pipe.

This is built on the architecture ideals that if you want to send something down the pipe, then this class can return you the response just fine. However, if you wish to receive content for this class then you need to register an async method to handle that incoming data as this class doesn’t have the context required to handle it.

async block_until_closed()[source]#

A blocking call which releases when the WS closes.

Notes

This will block even if the underlying WS has yet to connect.

async connect_client(url: str, idp: IdentifyDataPacket) → None[source]#

A blocking call which connects the underlying WS.

async connect_server() → None[source]#
async close() → None[source]#

Close the underlying WS

async send(data: dict) → Future[source]#

Send data down the pipe and receive a response.

Parameters:

data (dict) – The packet to send down the pipe

Returns:

This future will contain the returned data once it becomes available

Return type:

asyncio.Future

async send_response(*, packet_id: str, data: Optional[dict] = None)[source]#

Sends a response to a received request without waiting anything.

Parameters:

  • packet_id (str) – The packet ID to respond to

  • data (dict) – The data being responded with

register_receiver(callback) → Router[source]#

Register the provided callback to handle packets with the request type.

Parameters:

callback – The callback to call with the packet data.

class zonis.RouteHandler[source]#

Bases: object

route(route_name: Optional[str] = None)[source]#

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.

load_routes() → None[source]#

Loads all decorated routes.

register_class_instance_for_routes(instance, *routes) → None[source]#

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.