Skip to content

BaseProtocolServer

hololinked.core.interfaces.protocol_server.BaseProtocolServer

Bases: Parameterized

Base class for protocol specific servers.

Subclass from here to implement a new protocol.

Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
class BaseProtocolServer(Parameterized):
    """
    Base class for protocol specific servers.

    Subclass from here to implement a new protocol.
    """

    id = String(default=None, allow_None=True)
    """Unique identifier for the server"""

    port = Integer(default=9000, bounds=(1, 65535))
    """The protocol port"""

    logger = ClassSelector(
        class_=(logging.Logger, structlog.stdlib.BoundLoggerBase),
        default=None,
        allow_None=True,
    )  # type: logging.Logger | structlog.stdlib.BoundLogger
    """Logger instance"""

    things: MutableMapping[str, Thing]
    """Every served `Thing`, by id. Sub-things are not served."""

    def __init__(self, things: list[Thing] | dict[str, Thing] | None = None, **kwargs) -> None:
        from hololinked.core.thing import Thing

        self.config: BaseModel | None = None
        self.things = TypeConstrainedDict({}, key_type=str, item_type=Thing)
        super().__init__(**kwargs)
        self.add_things(*(things.values() if isinstance(things, dict) else things or []))

    @classmethod
    def from_params(cls, id: str, params: str | int | dict | list[str] | None) -> Self:
        """
        Create a server from the parameters given for its access point.

        Each protocol accepts some shorthand arguments, such as a port for HTTP, a broker hostname for MQTT etc.,
        which this method normalizes into constructor arguments.

        Parameters
        ----------
        id: str
            identifier for the server, used by the protocols that need one for routing
        params: str | int | dict | list[str] | None
            the parameters given for this protocol, either a protocol specific shorthand or a dict of keyword
            arguments for the constructor

        Returns
        -------
        Self
            the server, with no `Thing` added to it yet

        Raises
        ------
        ValueError
            if the parameters are not of a type this protocol supports
        """
        if not isinstance(params, dict):
            raise ValueError(f"{cls.__name__} parameters must be supplied as a dict, given : {type(params)}")
        return cls(**params)

    def add_thing(self, thing: Thing) -> None:
        """
        Adds a thing to the things being served.

        Sub-things are not served - see `EventLoop.add_thing`, which does not register them
        either.
        """
        self.things[thing.id] = thing

    def add_things(self, *things: Thing) -> None:
        """Adds multiple things to be served."""
        for thing in things:
            self.add_thing(thing)

    def add_property(self, *args, **kwargs) -> None:
        """
        Add a property to be served.

        Raises
        ------
        NotImplementedError
            if the protocol does not support this operation
        """
        raise NotImplementedError("Not implemented for this protocol")

    def add_action(self, *args, **kwargs) -> None:
        """
        Add an action to be served.

        Raises
        ------
        NotImplementedError
            if the protocol does not support this operation
        """
        raise NotImplementedError("Not implemented for this protocol")

    def add_event(self, *args, **kwargs) -> None:
        """
        Add an event to be served.

        Raises
        ------
        NotImplementedError
            if the protocol does not support this operation
        """
        raise NotImplementedError("Not implemented for this protocol")

    async def setup(self) -> None:
        """
        Prepare the protocol before it starts serving, creating side effects only without blocking.

        Raises
        ------
        NotImplementedError
            if the protocol does not implement a setup step
        """
        # This method should not block, just create side-effects
        raise NotImplementedError("Not implemented for this protocol")

    async def start(self) -> None:
        """
        Start serving the protocol, creating side effects only without blocking.

        Raises
        ------
        NotImplementedError
            if the protocol cannot be started this way
        """
        # This method should not block, just create side-effects
        # await self.setup()  # usually one should call setup() here
        raise NotImplementedError("Not implemented for this protocol")

    def welcome_lines(self) -> list[str]:
        """
        Lines announcing where this protocol can be reached (string), printed when a run starts.

        Returns
        -------
        list[str]
            the lines to print, without trailing newlines
        """
        return []

    @forkable
    def run(self, forked: bool = False, print_welcome_message: bool = True) -> None:
        """
        Run the server and serve your things.

        Use this method if this is the only running protocol. Blocks.

        Parameters
        ----------
        forked: bool, default False
            whether to run in a forked thread
        print_welcome_message: bool, default True
            whether to print a welcome message on startup, like the ports and access points
        """
        from hololinked.server import run

        run(self, print_welcome_message=print_welcome_message)

    def stop(self):
        """
        Stop serving the protocol.

        Stops this protocol only, the Thing still keeps running. To stop completely:

        ```python
        from hololinked.server import stop
        stop()
        ```

        Raises
        ------
        NotImplementedError
            if the protocol does not implement a stop step
        """
        raise NotImplementedError("Not implemented for this protocol")

Attributes

id class-attribute instance-attribute 

id = String(default=None, allow_None=True)

Unique identifier for the server

port class-attribute instance-attribute 

port = Integer(default=9000, bounds=(1, 65535))

The protocol port

logger class-attribute instance-attribute 

logger = ClassSelector(class_=(Logger, BoundLoggerBase), default=None, allow_None=True)

Logger instance

config instance-attribute 

config: BaseModel | None = None

things instance-attribute 

things: MutableMapping[str, Thing] = TypeConstrainedDict({}, key_type=str, item_type=Thing)

Every served Thing, by id. Sub-things are not served.

Functions

__init__ 

__init__(things: list[Thing] | dict[str, Thing] | None = None, **kwargs) -> None
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
46
47
48
49
50
51
52
def __init__(self, things: list[Thing] | dict[str, Thing] | None = None, **kwargs) -> None:
    from hololinked.core.thing import Thing

    self.config: BaseModel | None = None
    self.things = TypeConstrainedDict({}, key_type=str, item_type=Thing)
    super().__init__(**kwargs)
    self.add_things(*(things.values() if isinstance(things, dict) else things or []))

from_params classmethod 

from_params(id: str, params: str | int | dict | list[str] | None) -> Self

Create a server from the parameters given for its access point.

Each protocol accepts some shorthand arguments, such as a port for HTTP, a broker hostname for MQTT etc., which this method normalizes into constructor arguments.

Parameters:

Name Type Description Default
id
str

identifier for the server, used by the protocols that need one for routing

 required
params
str | int | dict | list[str] | None

the parameters given for this protocol, either a protocol specific shorthand or a dict of keyword arguments for the constructor

 required

Returns:

Type Description
Self

the server, with no Thing added to it yet

Raises:

Type Description
ValueError

if the parameters are not of a type this protocol supports
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@classmethod
def from_params(cls, id: str, params: str | int | dict | list[str] | None) -> Self:
    """
    Create a server from the parameters given for its access point.

    Each protocol accepts some shorthand arguments, such as a port for HTTP, a broker hostname for MQTT etc.,
    which this method normalizes into constructor arguments.

    Parameters
    ----------
    id: str
        identifier for the server, used by the protocols that need one for routing
    params: str | int | dict | list[str] | None
        the parameters given for this protocol, either a protocol specific shorthand or a dict of keyword
        arguments for the constructor

    Returns
    -------
    Self
        the server, with no `Thing` added to it yet

    Raises
    ------
    ValueError
        if the parameters are not of a type this protocol supports
    """
    if not isinstance(params, dict):
        raise ValueError(f"{cls.__name__} parameters must be supplied as a dict, given : {type(params)}")
    return cls(**params)

add_thing 

add_thing(thing: Thing) -> None

Adds a thing to the things being served.

Sub-things are not served - see EventLoop.add_thing, which does not register them either.

Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
84
85
86
87
88
89
90
91
def add_thing(self, thing: Thing) -> None:
    """
    Adds a thing to the things being served.

    Sub-things are not served - see `EventLoop.add_thing`, which does not register them
    either.
    """
    self.things[thing.id] = thing

add_things 

add_things(*things: Thing) -> None

Adds multiple things to be served.

Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
93
94
95
96
def add_things(self, *things: Thing) -> None:
    """Adds multiple things to be served."""
    for thing in things:
        self.add_thing(thing)

add_property 

add_property(*args, **kwargs) -> None

Add a property to be served.

Raises:

Type Description
NotImplementedError

if the protocol does not support this operation
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
 98
 99
100
101
102
103
104
105
106
107
def add_property(self, *args, **kwargs) -> None:
    """
    Add a property to be served.

    Raises
    ------
    NotImplementedError
        if the protocol does not support this operation
    """
    raise NotImplementedError("Not implemented for this protocol")

add_action 

add_action(*args, **kwargs) -> None

Add an action to be served.

Raises:

Type Description
NotImplementedError

if the protocol does not support this operation
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
109
110
111
112
113
114
115
116
117
118
def add_action(self, *args, **kwargs) -> None:
    """
    Add an action to be served.

    Raises
    ------
    NotImplementedError
        if the protocol does not support this operation
    """
    raise NotImplementedError("Not implemented for this protocol")

add_event 

add_event(*args, **kwargs) -> None

Add an event to be served.

Raises:

Type Description
NotImplementedError

if the protocol does not support this operation
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
120
121
122
123
124
125
126
127
128
129
def add_event(self, *args, **kwargs) -> None:
    """
    Add an event to be served.

    Raises
    ------
    NotImplementedError
        if the protocol does not support this operation
    """
    raise NotImplementedError("Not implemented for this protocol")

setup async 

setup() -> None

Prepare the protocol before it starts serving, creating side effects only without blocking.

Raises:

Type Description
NotImplementedError

if the protocol does not implement a setup step
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
131
132
133
134
135
136
137
138
139
140
141
async def setup(self) -> None:
    """
    Prepare the protocol before it starts serving, creating side effects only without blocking.

    Raises
    ------
    NotImplementedError
        if the protocol does not implement a setup step
    """
    # This method should not block, just create side-effects
    raise NotImplementedError("Not implemented for this protocol")

start async 

start() -> None

Start serving the protocol, creating side effects only without blocking.

Raises:

Type Description
NotImplementedError

if the protocol cannot be started this way
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
143
144
145
146
147
148
149
150
151
152
153
154
async def start(self) -> None:
    """
    Start serving the protocol, creating side effects only without blocking.

    Raises
    ------
    NotImplementedError
        if the protocol cannot be started this way
    """
    # This method should not block, just create side-effects
    # await self.setup()  # usually one should call setup() here
    raise NotImplementedError("Not implemented for this protocol")

run 

run(forked: bool = False, print_welcome_message: bool = True) -> None

Run the server and serve your things.

Use this method if this is the only running protocol. Blocks.

Parameters:

Name Type Description Default
forked
bool

whether to run in a forked thread

 False
print_welcome_message
bool

whether to print a welcome message on startup, like the ports and access points

 True
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@forkable
def run(self, forked: bool = False, print_welcome_message: bool = True) -> None:
    """
    Run the server and serve your things.

    Use this method if this is the only running protocol. Blocks.

    Parameters
    ----------
    forked: bool, default False
        whether to run in a forked thread
    print_welcome_message: bool, default True
        whether to print a welcome message on startup, like the ports and access points
    """
    from hololinked.server import run

    run(self, print_welcome_message=print_welcome_message)

stop 

stop()

Stop serving the protocol.

Stops this protocol only, the Thing still keeps running. To stop completely:

from hololinked.server import stop
stop()

Raises:

Type Description
NotImplementedError

if the protocol does not implement a stop step
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def stop(self):
    """
    Stop serving the protocol.

    Stops this protocol only, the Thing still keeps running. To stop completely:

    ```python
    from hololinked.server import stop
    stop()
    ```

    Raises
    ------
    NotImplementedError
        if the protocol does not implement a stop step
    """
    raise NotImplementedError("Not implemented for this protocol")

welcome_lines 

welcome_lines() -> list[str]

Lines announcing where this protocol can be reached (string), printed when a run starts.

Returns:

Type Description
list[str]

the lines to print, without trailing newlines
Source code in repo/hololinked/hololinked/core/interfaces/protocol_server.py 
156
157
158
159
160
161
162
163
164
165
def welcome_lines(self) -> list[str]:
    """
    Lines announcing where this protocol can be reached (string), printed when a run starts.

    Returns
    -------
    list[str]
        the lines to print, without trailing newlines
    """
    return []

The contract every protocol server must satisfy.