:py:mod:`aiosignald.util`
=========================

.. py:module:: aiosignald.util


Module Contents
---------------

Classes
~~~~~~~

.. autoapisummary::

   aiosignald.util.Handler
   aiosignald.util.JSONProtocol



Functions
~~~~~~~~~

.. autoapisummary::

   aiosignald.util.locals_to_request
   aiosignald.util.evaluate_string_or_forward_ref
   aiosignald.util.dict_to_nested_dataclass
   aiosignald.util.raise_if_needed



Attributes
~~~~~~~~~~

.. autoapisummary::

   aiosignald.util.T
   aiosignald.util._FIELDS


.. py:exception:: SignaldException(payload)

   Bases: :py:obj:`Exception`

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

   .. py:method:: __str__()

      Return str(self).



.. py:exception:: SignaldUnlistedException(payload)

   Bases: :py:obj:`SignaldException`

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


.. py:class:: 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).

   :param validator: a Callable that will receive the response and return True
       if the response is the one it was waiting for
   :param callback: a Callable that will be called with the response as argument

   .. py:attribute:: validator
      :type: Callable

      

   .. py:attribute:: callback
      :type: Callable

      

   .. py:method:: validate(response)



.. py:class:: JSONProtocol(on_con_lost: Optional[asyncio.Future] = None)

   Bases: :py:obj:`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()

   .. py:attribute:: PROTOCOL_VERSION
      :value: 'v1'

      

   .. py:method:: 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.


   .. py:method:: connection_lost(exc: Union[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).


   .. py:method:: 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}.



   .. py:method:: __wrap(handler: Callable[[Any, dict], Awaitable[None]], data: Any, payload: dict)
      :async:


   .. py:method:: 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


   .. py:method:: send_request(payload: dict, id_: Optional[str] = None, callback: Optional[Callable] = None)

      Send a JSON payload to the UNIX socket.

      :param payload: dict
      :param id_: identifier of the request. If not specified, a random string is used.
      :param callback: a Callable that will be called with the response payload


   .. py:method:: get_response(payload: dict, full: bool = False) -> dict
      :async:

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

      :param payload:
      :param full: If False, return only payload["data"]


   .. py:method:: handle_version(_data, payload)


   .. py:method:: handle_JsonVersionMessage(version)



.. py:function:: locals_to_request(d: dict)

   Helper for the generated bindings.


.. py:data:: T

   

.. py:function:: evaluate_string_or_forward_ref(t: Any)


.. py:function:: dict_to_nested_dataclass(cls: Type[T], dikt) -> T


.. py:function:: raise_if_needed(response: dict)

   Raise a python exception using a signald response payload


.. py:data:: _FIELDS
   :value: '__dataclass_fields__'