pamoja.core
The pamoja engine's surface: the runtime version, the error every native call raises, and the transport every link shares.
This is the counterpart of the pamoja-core crate, and like it, it is small; the
compiled engine is pamoja-native, which this package depends on. Each capability
is its own package (pamoja-mqtt gives pamoja.mqtt, and so on), and
pamoja installs all of them.
A ladder rung, a fault injector, and a degraded link all take some transport. Python has no way to say "any transport", so one class holds whichever kind was built and dispatches to it. Composing consumes a transport, because the thing it is composed into owns it from then on; a consumed transport is emptied rather than left aliasing what now belongs to a ladder, so using one twice raises.
1"""The pamoja engine's surface: the runtime version, the error every native call 2raises, and the transport every link shares. 3 4This is the counterpart of the ``pamoja-core`` crate, and like it, it is small; the 5compiled engine is ``pamoja-native``, which this package depends on. Each capability 6is its own package (``pamoja-mqtt`` gives ``pamoja.mqtt``, and so on), and 7``pamoja`` installs all of them. 8 9A ladder rung, a fault injector, and a degraded link all take some transport. 10Python has no way to say "any transport", so one class holds whichever kind was 11built and dispatches to it. Composing consumes a transport, because the thing it 12is composed into owns it from then on; a consumed transport is emptied rather 13than left aliasing what now belongs to a ladder, so using one twice raises. 14""" 15 16from __future__ import annotations 17 18from typing import Awaitable, Protocol, Union 19 20from pamoja._native import Message, PamojaError, version 21from pamoja._native import PyTransport as Transport 22 23__all__ = [ 24 "Message", 25 "PamojaError", 26 "ReceivingTransportHandlers", 27 "Transport", 28 "TransportHandlers", 29 "version", 30] 31 32 33class TransportHandlers(Protocol): 34 """The methods a link written in Python supplies to stand as a transport. 35 36 Implement them for a link pamoja does not ship, a vendor SDK, a proprietary 37 radio, or a cloud client, and hand the object to ``Transport.from_handlers``. 38 Each may be a coroutine function or a plain one. One that raises fails the call 39 that reached it, with what it raised. 40 """ 41 42 def connect(self) -> Union[Awaitable[None], None]: 43 """Establish the link. Called again on every reconnect.""" 44 45 def send(self, topic: str, payload: bytes) -> Union[Awaitable[None], None]: 46 """Publish a payload to a topic.""" 47 48 def subscribe(self, topic: str) -> Union[Awaitable[None], None]: 49 """Subscribe to a topic filter. A ladder places its filters again on every reconnect.""" 50 51 52class ReceivingTransportHandlers(TransportHandlers, Protocol): 53 """A link that also delivers what it subscribed to. 54 55 ``recv`` is called again as soon as it returns, from the moment the transport 56 connects, and what it delivers is queued for the receiving side. One that raises 57 ends the link until the next connect, and the next receive raises what it raised. 58 """ 59 60 def recv( 61 self, 62 ) -> Union[ 63 Awaitable[Union[Message, tuple[str, Union[str, bytes]], None]], 64 Message, 65 tuple[str, Union[str, bytes]], 66 None, 67 ]: 68 """Wait for the next message: a ``Message``, a ``(topic, payload)`` pair, or 69 ``None`` once the link has ended."""
A message that arrived on a subscribed topic.
CoAP and the loopback broker both hand back a topic and a payload, so one class serves them rather than a near-identical type per transport.
Raised when a pamoja operation fails.
53class ReceivingTransportHandlers(TransportHandlers, Protocol): 54 """A link that also delivers what it subscribed to. 55 56 ``recv`` is called again as soon as it returns, from the moment the transport 57 connects, and what it delivers is queued for the receiving side. One that raises 58 ends the link until the next connect, and the next receive raises what it raised. 59 """ 60 61 def recv( 62 self, 63 ) -> Union[ 64 Awaitable[Union[Message, tuple[str, Union[str, bytes]], None]], 65 Message, 66 tuple[str, Union[str, bytes]], 67 None, 68 ]: 69 """Wait for the next message: a ``Message``, a ``(topic, payload)`` pair, or 70 ``None`` once the link has ended."""
A link that also delivers what it subscribed to.
recv is called again as soon as it returns, from the moment the transport
connects, and what it delivers is queued for the receiving side. One that raises
ends the link until the next connect, and the next receive raises what it raised.
61 def recv( 62 self, 63 ) -> Union[ 64 Awaitable[Union[Message, tuple[str, Union[str, bytes]], None]], 65 Message, 66 tuple[str, Union[str, bytes]], 67 None, 68 ]: 69 """Wait for the next message: a ``Message``, a ``(topic, payload)`` pair, or 70 ``None`` once the link has ended."""
Wait for the next message: a Message, a (topic, payload) pair, or
None once the link has ended.
Inherited Members
34class TransportHandlers(Protocol): 35 """The methods a link written in Python supplies to stand as a transport. 36 37 Implement them for a link pamoja does not ship, a vendor SDK, a proprietary 38 radio, or a cloud client, and hand the object to ``Transport.from_handlers``. 39 Each may be a coroutine function or a plain one. One that raises fails the call 40 that reached it, with what it raised. 41 """ 42 43 def connect(self) -> Union[Awaitable[None], None]: 44 """Establish the link. Called again on every reconnect.""" 45 46 def send(self, topic: str, payload: bytes) -> Union[Awaitable[None], None]: 47 """Publish a payload to a topic.""" 48 49 def subscribe(self, topic: str) -> Union[Awaitable[None], None]: 50 """Subscribe to a topic filter. A ladder places its filters again on every reconnect."""
The methods a link written in Python supplies to stand as a transport.
Implement them for a link pamoja does not ship, a vendor SDK, a proprietary
radio, or a cloud client, and hand the object to Transport.from_handlers.
Each may be a coroutine function or a plain one. One that raises fails the call
that reached it, with what it raised.
1968def _no_init_or_replace_init(self, *args, **kwargs): 1969 cls = type(self) 1970 1971 if cls._is_protocol: 1972 raise TypeError('Protocols cannot be instantiated') 1973 1974 # Already using a custom `__init__`. No need to calculate correct 1975 # `__init__` to call. This can lead to RecursionError. See bpo-45121. 1976 if cls.__init__ is not _no_init_or_replace_init: 1977 return 1978 1979 # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`. 1980 # The first instantiation of the subclass will call `_no_init_or_replace_init` which 1981 # searches for a proper new `__init__` in the MRO. The new `__init__` 1982 # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent 1983 # instantiation of the protocol subclass will thus use the new 1984 # `__init__` and no longer call `_no_init_or_replace_init`. 1985 for base in cls.__mro__: 1986 init = base.__dict__.get('__init__', _no_init_or_replace_init) 1987 if init is not _no_init_or_replace_init: 1988 cls.__init__ = init 1989 break 1990 else: 1991 # should not happen 1992 cls.__init__ = object.__init__ 1993 1994 cls.__init__(self, *args, **kwargs)
43 def connect(self) -> Union[Awaitable[None], None]: 44 """Establish the link. Called again on every reconnect."""
Establish the link. Called again on every reconnect.
46 def send(self, topic: str, payload: bytes) -> Union[Awaitable[None], None]: 47 """Publish a payload to a topic."""
Publish a payload to a topic.
49 def subscribe(self, topic: str) -> Union[Awaitable[None], None]: 50 """Subscribe to a topic filter. A ladder places its filters again on every reconnect."""
Subscribe to a topic filter. A ladder places its filters again on every reconnect.
Returns the version of the native pamoja module.