aiosignald.util

Module Contents

Classes

Handler

Allows to await for a specific message sent by JSONProtocol.

JSONProtocol

Interface for stream protocol.

Functions

locals_to_request(d)

Helper for the generated bindings.

evaluate_string_or_forward_ref(t)

dict_to_nested_dataclass(→ T)

raise_if_needed(response)

Raise a python exception using a signald response payload

Attributes

T

_FIELDS

exception aiosignald.util.SignaldException(payload)

Bases: Exception

Base class to translate signald’s response payloads into python exceptions

__str__()

Return str(self).

exception aiosignald.util.SignaldUnlistedException(payload)

Bases: SignaldException

Base class to translate signald’s response payloads into python exceptions

class aiosignald.util.Handler

Allows to await for a specific message sent by JSONProtocol. This should not be used directly, but rather using JSONProtocol.get_future_for(validator).

Parameters:

  • validator – a Callable that will receive the response and return True if the response is the one it was waiting for

  • callback – a Callable that will be called with the response as argument

validator: Callable
callback: Callable
validate(response)
class aiosignald.util.JSONProtocol(on_con_lost: asyncio.Future | None = None)

Bases: asyncio.Protocol

Interface for stream protocol.

The user should implement this interface. They can inherit from this class but don’t need to. The implementations here do nothing (they don’t raise exceptions).

When the user wants to requests a transport, they pass a protocol factory to a utility function (e.g., EventLoop.create_connection()).

When the connection is made successfully, connection_made() is called with a suitable transport object. Then data_received() will be called 0 or more times with data (bytes) received from the transport; finally, connection_lost() will be called exactly once with either an exception object or None as an argument.

State machine of calls:

start -> CM [-> DR*] [-> ER?] -> CL -> end

  • CM: connection_made()

  • DR: data_received()

  • ER: eof_received()

  • CL: connection_lost()

PROTOCOL_VERSION = 'v1'
connection_made(transport: asyncio.transports.BaseTransport)

Called when a connection is made.

The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.

connection_lost(exc: Exception | None)

Called when the connection is lost or closed.

The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed).

data_received(data: bytes)

Handle data received through the UNIX socket, convert it to a python dict and dispatch it to the relevant callback or handler.

Callbacks can either:

  • wait for a specific payload id, for this you should use the coroutine JSONProtocol.get_response

  • wait for a more generic matching payload, for this you should use the coroutine JSONProtocol.get_future_for

In case no callbacks match the payload, it is sent to the method JSONProtocol.handle_{payload.type}.

async __wrap(handler: Callable[[Any, dict], Awaitable[None]], data: Any, payload: dict)
exception(handler: Callable, e: Exception)

Handle exceptions in generic handlers

Override this function or set the attribute to customize the default behaviour, which is to just log the exception

send_request(payload: dict, id_: str | None = None, callback: Callable | None = None)

Send a JSON payload to the UNIX socket.

Parameters:

  • payload – dict

  • id – identifier of the request. If not specified, a random string is used.

  • callback – a Callable that will be called with the response payload

async get_response(payload: dict, full: bool = False) → dict

Coroutine to await the response to a specific payload. Can raise exceptions in case the response is an error.

Parameters:

  • payload –

  • full – If False, return only payload[“data”]

handle_version(_data, payload)
handle_JsonVersionMessage(version)
aiosignald.util.locals_to_request(d: dict)

Helper for the generated bindings.

aiosignald.util.T
aiosignald.util.evaluate_string_or_forward_ref(t: Any)
aiosignald.util.dict_to_nested_dataclass(cls: Type[T], dikt) → T
aiosignald.util.raise_if_needed(response: dict)

Raise a python exception using a signald response payload

aiosignald.util._FIELDS = '__dataclass_fields__'