ProtocolServers

hololinked.injection.ProtocolServers

Bases: Registry

A singleton registry of the protocol servers a Thing can be exposed with.

All members are class attributes and settings are applied process-wide (python process).

from hololinked import ProtocolServers

ProtocolServers.register(WebsocketServerWithMyCustomProtocol, "websockets", protocol="websockets")

thing.run(access_points=[("websockets", 5683)])

# OR

WebsocketServerWithMyCustomProtocol.add_thing(thing)
WebsocketServerWithMyCustomProtocol.run()

Source code in repo/hololinked/hololinked/injection.py

class ProtocolServers(Registry):
    """
    A singleton registry of the protocol servers a `Thing` can be exposed with.

    All members are class attributes and settings are applied process-wide (python process).

    ```python
    from hololinked import ProtocolServers

    ProtocolServers.register(WebsocketServerWithMyCustomProtocol, "websockets", protocol="websockets")

    thing.run(access_points=[("websockets", 5683)])

    # OR

    WebsocketServerWithMyCustomProtocol.add_thing(thing)
    WebsocketServerWithMyCustomProtocol.run()
    ```
    """

    adapter_kind: ClassVar[str] = "protocol server"
    package: ClassVar[str] = "hololinked.server"
    tables: ClassVar[tuple[str, ...]] = ("modules", "protocols")

    modules: ClassVar[dict[str, str | tuple[str, str]]] = {
        "http": "HTTPServer",
        "mqtt": "MQTTPublisher",
        "zmq": "ZMQServer",
    }

    http: type[BaseProtocolServer]
    """HTTP(s) server exposing `Thing`s over HTTP 1.1."""
    mqtt: type[BaseProtocolServer]
    """MQTT publisher pushing events and observable properties to a broker's topic tree."""
    zmq: type[BaseProtocolServer]
    """ZeroMQ server exposing `Thing`s over IPC, TCP and INPROC transport."""

    protocols: ClassVar[dict[str, str]] = {
        "HTTP": "http",
        "MQTT": "mqtt",
        "ZMQ": "zmq",
    }
    """
    Protocol name, as given in an access point, mapped to the server that serves it.

    Answerable without importing the server, which is what lets an access point select one lazily.
    """

    @classmethod
    def register(cls, server: type[BaseProtocolServer], name: str, protocol: str) -> None:
        """
        Register a protocol server under a given name.

        Parameters
        ----------
        server: type[BaseProtocolServer]
            the server class to register, must be a subclass of `BaseProtocolServer`
        name: str
            the name to register the server under, for example 'http' or 'zmq'
        protocol: str
            the protocol name that selects this server in an access point, for example 'HTTP'. Case insensitive.

        Raises
        ------
        TypeError
            if the server is not a subclass of `BaseProtocolServer`
        """
        if not issubklass(server, BaseProtocolServer):
            raise TypeError(f"server must be a subclass of BaseProtocolServer, given : {server}")
        cls.install(name, server)
        cls.protocols[protocol.upper()] = name

    @classmethod
    def for_protocol(cls, protocol: str) -> type[BaseProtocolServer] | None:
        """
        Get the server class that serves a given protocol, importing it if necessary.

        Parameters
        ----------
        protocol: str
            the protocol name as given in an access point, for example 'HTTP'. Case insensitive.

        Returns
        -------
        type[BaseProtocolServer] | None
            the server class, to be created with its `from_params()`, None if no server serves this protocol
        """
        name = cls.protocols.get(protocol.upper())
        return None if name is None else getattr(cls, name)

    @classmethod
    def reset(cls) -> None:
        """Reset the protocol server registry."""
        cls.forget_adapters()

Attributes

http instance-attribute

http: type[BaseProtocolServer]

HTTP(s) server exposing Things over HTTP 1.1.

mqtt instance-attribute

mqtt: type[BaseProtocolServer]

MQTT publisher pushing events and observable properties to a broker's topic tree.

zmq instance-attribute

zmq: type[BaseProtocolServer]

ZeroMQ server exposing Things over IPC, TCP and INPROC transport.

modules class-attribute

modules: dict[str, str | tuple[str, str]] = {'http': 'HTTPServer', 'mqtt': 'MQTTPublisher', 'zmq': 'ZMQServer'}

protocols class-attribute

protocols: dict[str, str] = {'HTTP': 'http', 'MQTT': 'mqtt', 'ZMQ': 'zmq'}

Protocol name, as given in an access point, mapped to the server that serves it.

Answerable without importing the server, which is what lets an access point select one lazily.

Functions

for_protocol classmethod

for_protocol(protocol: str) -> type[BaseProtocolServer] | None

Get the server class that serves a given protocol, importing it if necessary.

Parameters:

Name	Type	Description	Default
protocol	str	the protocol name as given in an access point, for example 'HTTP'. Case insensitive.	required

Returns:

Type	Description
type[BaseProtocolServer] | None	the server class, to be created with its from_params(), None if no server serves this protocol

Source code in repo/hololinked/hololinked/injection.py

@classmethod
def for_protocol(cls, protocol: str) -> type[BaseProtocolServer] | None:
    """
    Get the server class that serves a given protocol, importing it if necessary.

    Parameters
    ----------
    protocol: str
        the protocol name as given in an access point, for example 'HTTP'. Case insensitive.

    Returns
    -------
    type[BaseProtocolServer] | None
        the server class, to be created with its `from_params()`, None if no server serves this protocol
    """
    name = cls.protocols.get(protocol.upper())
    return None if name is None else getattr(cls, name)

register classmethod

register(server: type[BaseProtocolServer], name: str, protocol: str) -> None

Register a protocol server under a given name.

Parameters:

Name	Type	Description	Default
server	type[BaseProtocolServer]	the server class to register, must be a subclass of BaseProtocolServer	required
name	str	the name to register the server under, for example 'http' or 'zmq'	required
protocol	str	the protocol name that selects this server in an access point, for example 'HTTP'. Case insensitive.	required

Raises:

Type	Description
TypeError	if the server is not a subclass of BaseProtocolServer

Source code in repo/hololinked/hololinked/injection.py

@classmethod
def register(cls, server: type[BaseProtocolServer], name: str, protocol: str) -> None:
    """
    Register a protocol server under a given name.

    Parameters
    ----------
    server: type[BaseProtocolServer]
        the server class to register, must be a subclass of `BaseProtocolServer`
    name: str
        the name to register the server under, for example 'http' or 'zmq'
    protocol: str
        the protocol name that selects this server in an access point, for example 'HTTP'. Case insensitive.

    Raises
    ------
    TypeError
        if the server is not a subclass of `BaseProtocolServer`
    """
    if not issubklass(server, BaseProtocolServer):
        raise TypeError(f"server must be a subclass of BaseProtocolServer, given : {server}")
    cls.install(name, server)
    cls.protocols[protocol.upper()] = name

reset classmethod

reset() -> None

Reset the protocol server registry.

Source code in repo/hololinked/hololinked/injection.py

@classmethod
def reset(cls) -> None:
    """Reset the protocol server registry."""
    cls.forget_adapters()