Skip to content

hololinked.server.http.HTTPServer

Bases: BaseProtocolServer

HTTP(s) server to expose Thing over HTTP protocol. Supports HTTP 1.1. Use add_thing, or add_property or add_action or add_event methods to add things to the server.

Source code in hololinked/hololinked/server/http/server.py 
 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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
class HTTPServer(BaseProtocolServer):
    """
    HTTP(s) server to expose `Thing` over HTTP protocol. Supports HTTP 1.1.
    Use `add_thing`, or `add_property` or `add_action` or `add_event` methods to add things to the server.
    """

    address = IPAddress(default="0.0.0.0", doc="IP address")  # type: str
    # SAST(id='hololinked.server.http.HTTPServer.address', description='B104:hardcoded_bind_all_interfaces', tool='bandit')
    """IP address, especially to bind to all interfaces or not"""

    ssl_context = ClassSelector(
        class_=ssl.SSLContext,
        default=None,
        allow_None=True,
    )  # type: ssl.SSLContext | None
    """SSL context to provide encrypted communication"""

    config = ClassSelector(
        class_=RuntimeConfig,
        default=None,
        allow_None=True,
    )  # type: RuntimeConfig
    """Runtime configuration for the HTTP server. See `hololinked.server.http.config.RuntimeConfig` for details"""

    def __init__(
        self,
        *,
        port: int = 8080,
        address: str = "0.0.0.0",  # SAST(id='hololinked.server.http.HTTPServer.__init__.address', description='B104:hardcoded_bind_all_interfaces', tool='bandit')
        things: list[Thing] | None = None,
        # host: Optional[str] = None,
        logger: structlog.stdlib.BoundLogger | None = None,
        ssl_context: ssl.SSLContext | None = None,
        security_schemes: list[Security] | None = None,
        # protocol_version : int = 1, network_interface : str = 'Ethernet',
        allowed_clients: str | Iterable[str] | None = None,
        config: dict[str, Any] | None = None,
        **kwargs,
    ) -> None:
        """
        Parameters
        ----------
        port: int, default 8080
            the port at which the server should be run
        address: str, default 0.0.0.0
            IP address, use 0.0.0.0 to bind to all interfaces to expose the server to other devices in the network
            and 127.0.0.1 to bind only to localhost
        logger: structlog.stdlib.BoundLogger, optional
            structlog.stdlib.BoundLogger instance
        ssl_context: ssl.SSLContext
            SSL context to provide encrypted communication
        security_schemes: list[Security], optional
            list of security schemes to be used by the server. If None, no security scheme is used.
        allowed_clients: List[str]
            serves request and sets CORS only from these clients, other clients are reject with 403. Unlike pure CORS
            feature, the server resource is not even executed if the client is not an allowed client.
        **kwargs:
            additional keyword arguments for server configuration. Usually:

            - `property_handler`: `BaseHandler` | `PropertyHandler`, optional.
                custom web request handler for property read-write
            - `action_handler`: `BaseHandler` | `ActionHandler`, optional.
                custom web request handler for action
            - `event_handler`: `EventHandler` | `BaseHandler`, optional.
                custom event handler for sending HTTP SSE

            or RuntimeConfig attributes can be passed as keyword arguments.
        """
        default_config = dict(
            cors=global_config.ALLOW_CORS,
            property_handler=kwargs.get("property_handler", PropertyHandler),
            action_handler=kwargs.get("action_handler", ActionHandler),
            event_handler=kwargs.get("event_handler", EventHandler),
            thing_description_handler=kwargs.get("thing_description_handler", ThingDescriptionHandler),
            RW_multiple_properties_handler=kwargs.get("RW_multiple_properties_handler", RWMultiplePropertiesHandler),
            liveness_probe_handler=kwargs.get("liveness_handler", LivenessProbeHandler),
            readiness_probe_handler=kwargs.get("readiness_handler", ReadinessProbeHandler),
            stop_handler=kwargs.get("stop_handler", StopHandler),
            thing_description_service=kwargs.get("thing_description_service", ThingDescriptionService),
            thing_repository=kwargs.get("thing_repository", dict()),
            allowed_clients=allowed_clients,
            security_schemes=security_schemes,
        )
        default_config.update(config or dict())
        config = RuntimeConfig(**default_config)
        # need to be extended when more options are added
        super().__init__(
            port=port,
            address=address,
            logger=logger,
            ssl_context=ssl_context,
            config=config,
        )

        self._IP = f"{self.address}:{self.port}"  # TODO, remove this variable later?
        self.id = self._IP
        if self.logger is None:
            self.logger = structlog.get_logger().bind(component="http-server", host=f"{self.address}:{self.port}")

        self.tornado_instance = None
        self.app = Application(
            handlers=[
                (
                    r"/stop",
                    self.config.stop_handler,
                    dict(config=self.config, logger=self.logger, owner_inst=self),
                ),
                (
                    r"/liveness",
                    self.config.liveness_probe_handler,
                    dict(config=self.config, logger=self.logger, owner_inst=self),
                ),
                (
                    r"/readiness",
                    self.config.readiness_probe_handler,
                    dict(config=self.config, logger=self.logger, owner_inst=self),
                ),
            ]
        )
        self.router = ApplicationRouter(self.app, self)

        self.zmq_client_pool = MessageMappedZMQClientPool(
            id=self.id,
            server_ids=[],
            client_ids=[],
            handshake=False,
            poll_timeout=100,
        )
        self.add_things(*(things or []))

    def setup(self) -> None:
        """check if all the requirements are met before starting the server, auto invoked by listen()"""
        # Add only those code here that needs to be redone always before restarting the server.
        # One time creation attributes/activities must be in init

        # Comments are above the relevant lines, not below
        # 1. clear the event loop in case any pending tasks exist, also restarting with same
        # event loop is buggy, so we remove it.
        ioloop.IOLoop.clear_current()
        # 2. sets async loop for a non-possessing thread as well
        event_loop = get_current_async_loop()
        # 3. schedule the ZMQ client pool polling
        event_loop.create_task(self.zmq_client_pool.poll_responses())
        # self.zmq_client_pool.handshake(), NOTE - handshake better done upfront as we already poll_responses here
        # which will prevent handshake function to succeed (although handshake will be done)
        # 4. Expose via broker
        for thing in self.things:
            if not thing.rpc_server:
                raise ValueError(f"You need to expose thing {thing.id} via a RPCServer before trying to serve it")
            event_loop.create_task(
                self._instantiate_broker(
                    thing.rpc_server.id,
                    thing.id,
                    "INPROC",
                )
            )
        # 5. finally also get a reference of the same event loop from tornado
        self.tornado_event_loop = ioloop.IOLoop.current()

        self.tornado_instance = TornadoHTTP1Server(self.app, ssl_options=self.ssl_context)  # type: TornadoHTTP1Server

    async def start(self) -> None:
        self.setup()
        self.tornado_instance.listen(port=self.port, address=self.address)
        self.logger.info(f"started HTTP webserver at {self._IP}, ready to receive requests.")

    def stop(self, attempt_async_stop: bool = True) -> None:
        """
        Stop the HTTP server - unreliable, use `async_stop()` if possible.
        A stop handler at the path `/stop` with POST method is already implemented that invokes this
        method for the clients.

        Parameters
        ----------
        attempt_async_stop: bool, default `True`
            if `True`, attempts to run the `async_stop` method to close all connections gracefully.
        """
        if attempt_async_stop:
            run_callable_somehow(self.async_stop())
            return
        self.zmq_client_pool.stop_polling()
        if not self.tornado_instance:
            return
        self.tornado_instance.stop()
        run_callable_somehow(self.tornado_instance.close_all_connections())

    async def async_stop(self) -> None:
        """
        Stop the HTTP server. A stop handler at the path `/stop` with POST method is already implemented
        that invokes this method for the clients.
        """
        self.zmq_client_pool.stop_polling()
        if not self.tornado_instance:
            return
        try:
            self.tornado_instance.stop()
            await self.tornado_instance.close_all_connections()
        except Exception as ex:
            self.logger.error(
                "error while stopping tornado server, use stop() method "
                + f"from hololinked.server and do not reuse the port - {ex}"
            )

    def add_property(
        self,
        URL_path: str,
        property: Property | PropertyAffordance,
        http_methods: str | tuple[str, str, str] = ("GET", "PUT"),
        handler: BaseHandler | PropertyHandler = PropertyHandler,
        **kwargs,
    ) -> None:
        """
        Add a property to be accessible by HTTP.

        Parameters
        ----------
        URL_path: str
            URL path to access the property
        property: Property | PropertyAffordance
            Property (object) to be served or its JSON representation
        http_methods: Tuple[str, str, str]
            tuple of http methods to be used for read, write and delete. Use None or omit HTTP method for
            unsupported operations. For example - for readonly property use ('GET', None, None) or ('GET',)
        handler: BaseHandler | PropertyHandler, optional
            custom handler for the property, otherwise the default handler will be used
        kwargs: dict
            additional keyword arguments to be passed to the handler's __init__
        """
        if not isinstance(property, (Property, PropertyAffordance)):
            raise TypeError(f"property should be of type Property, given type {type(property)}")
        if not issubklass(handler, BaseHandler):
            raise TypeError(f"handler should be subclass of BaseHandler, given type {type(handler)}")
        if isinstance(property, Property):
            property = property.to_affordance()
        read_http_method = write_http_method = delete_http_method = None
        http_methods = self.router.adapt_http_methods(http_methods)
        if len(http_methods) == 1:
            read_http_method = http_methods[0]
        elif len(http_methods) == 2:
            read_http_method, write_http_method = http_methods
        elif len(http_methods) == 3:
            read_http_method, write_http_method, delete_http_method = http_methods
        if read_http_method != "GET":
            raise ValueError("read method should be GET")
        if write_http_method and write_http_method not in ["POST", "PUT"]:
            raise ValueError("write method should be POST or PUT")
        if delete_http_method and delete_http_method != "DELETE":
            raise ValueError("delete method should be DELETE")
        kwargs["resource"] = property
        kwargs["logger"] = self.logger
        kwargs["config"] = self.config
        kwargs["metadata"] = HandlerMetadata(http_methods=http_methods)
        self.router.add_rule(affordance=property, URL_path=URL_path, handler=handler, kwargs=kwargs)

    def add_action(
        self,
        URL_path: str,
        action: Action | ActionAffordance,
        http_method: str | None = "POST",
        handler: BaseHandler | ActionHandler = ActionHandler,
        **kwargs,
    ) -> None:
        """
        Add an action to be accessible by HTTP

        Parameters
        ----------
        URL_path: str
            URL path to access the action
        action: Action | ActionAffordance
            Action (object) to be served or its JSON representation
        http_method: str
            http method to be used for the action
        handler: BaseHandler | ActionHandler, optional
            custom handler for the action
        kwargs: dict
            additional keyword arguments to be passed to the handler's __init__
        """
        if not isinstance(action, (Action, ActionAffordance)):
            raise TypeError(f"Given action should be of type Action or ActionAffordance, given type {type(action)}")
        if not issubklass(handler, BaseHandler):
            raise TypeError(f"handler should be subclass of BaseHandler, given type {type(handler)}")
        http_methods = self.router.adapt_http_methods(http_method)
        if isinstance(action, Action):
            action = action.to_affordance()  # type: ActionAffordance
        kwargs["resource"] = action
        kwargs["config"] = self.config
        kwargs["logger"] = self.logger
        kwargs["metadata"] = HandlerMetadata(http_methods=http_methods)
        self.router.add_rule(affordance=action, URL_path=URL_path, handler=handler, kwargs=kwargs)

    def add_event(
        self,
        URL_path: str,
        event: Event | EventAffordance | PropertyAffordance,
        handler: BaseHandler | EventHandler = EventHandler,
        **kwargs,
    ) -> None:
        """
        Add an event to be accessible by HTTP server; only GET method is supported for events.

        Parameters
        ----------
        URL_path: str
            URL path to access the event
        event: Event | EventAffordance
            Event (object) to be served or its JSON representation
        handler: BaseHandler | EventHandler, optional
            custom handler for the event
        kwargs: dict
            additional keyword arguments to be passed to the handler's __init__
        """
        if not isinstance(event, (Event, EventAffordance)) and (
            not isinstance(event, PropertyAffordance) or not event.observable
        ):
            raise TypeError(f"event should be of type Event or EventAffordance, given type {type(event)}")
        if not issubklass(handler, BaseHandler):
            raise TypeError(f"handler should be subclass of BaseHandler, given type {type(handler)}")
        if isinstance(event, Event):
            event = event.to_affordance()
        kwargs["resource"] = event
        kwargs["config"] = self.config
        kwargs["logger"] = self.logger
        kwargs["metadata"] = HandlerMetadata(http_methods=("GET",))
        self.router.add_rule(affordance=event, URL_path=URL_path, handler=handler, kwargs=kwargs)

    def add_thing(self, thing: Thing) -> None:
        self.router.add_thing(thing)
        self.things.append(thing)

    def __hash__(self):
        return hash(self._IP)

    def __eq__(self, other):
        if not isinstance(other, HTTPServer):
            return False
        return self._IP == other._IP

    def __str__(self):
        return f"{self.__class__.__name__}(address={self.address}, port={self.port})"

Functions

__init__ 

__init__(*, port: int = 8080, address: str = '0.0.0.0', things: list[Thing] | None = None, logger: BoundLogger | None = None, ssl_context: SSLContext | None = None, security_schemes: list[Security] | None = None, allowed_clients: str | Iterable[str] | None = None, config: dict[str, Any] | None = None, **kwargs) -> None

Parameters:

Name Type Description Default

port

int

the port at which the server should be run

 8080

address

str

IP address, use 0.0.0.0 to bind to all interfaces to expose the server to other devices in the network and 127.0.0.1 to bind only to localhost

 '0.0.0.0'

logger

BoundLogger | None

structlog.stdlib.BoundLogger instance

 None

ssl_context

SSLContext | None

SSL context to provide encrypted communication

 None

security_schemes

list[Security] | None

list of security schemes to be used by the server. If None, no security scheme is used.

 None

allowed_clients

str | Iterable[str] | None

serves request and sets CORS only from these clients, other clients are reject with 403. Unlike pure CORS feature, the server resource is not even executed if the client is not an allowed client.

 None

**kwargs

additional keyword arguments for server configuration. Usually:

  • property_handler: BaseHandler | PropertyHandler, optional. custom web request handler for property read-write
  • action_handler: BaseHandler | ActionHandler, optional. custom web request handler for action
  • event_handler: EventHandler | BaseHandler, optional. custom event handler for sending HTTP SSE

or RuntimeConfig attributes can be passed as keyword arguments.

 {}
Source code in hololinked/hololinked/server/http/server.py 
 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
def __init__(
    self,
    *,
    port: int = 8080,
    address: str = "0.0.0.0",  # SAST(id='hololinked.server.http.HTTPServer.__init__.address', description='B104:hardcoded_bind_all_interfaces', tool='bandit')
    things: list[Thing] | None = None,
    # host: Optional[str] = None,
    logger: structlog.stdlib.BoundLogger | None = None,
    ssl_context: ssl.SSLContext | None = None,
    security_schemes: list[Security] | None = None,
    # protocol_version : int = 1, network_interface : str = 'Ethernet',
    allowed_clients: str | Iterable[str] | None = None,
    config: dict[str, Any] | None = None,
    **kwargs,
) -> None:
    """
    Parameters
    ----------
    port: int, default 8080
        the port at which the server should be run
    address: str, default 0.0.0.0
        IP address, use 0.0.0.0 to bind to all interfaces to expose the server to other devices in the network
        and 127.0.0.1 to bind only to localhost
    logger: structlog.stdlib.BoundLogger, optional
        structlog.stdlib.BoundLogger instance
    ssl_context: ssl.SSLContext
        SSL context to provide encrypted communication
    security_schemes: list[Security], optional
        list of security schemes to be used by the server. If None, no security scheme is used.
    allowed_clients: List[str]
        serves request and sets CORS only from these clients, other clients are reject with 403. Unlike pure CORS
        feature, the server resource is not even executed if the client is not an allowed client.
    **kwargs:
        additional keyword arguments for server configuration. Usually:

        - `property_handler`: `BaseHandler` | `PropertyHandler`, optional.
            custom web request handler for property read-write
        - `action_handler`: `BaseHandler` | `ActionHandler`, optional.
            custom web request handler for action
        - `event_handler`: `EventHandler` | `BaseHandler`, optional.
            custom event handler for sending HTTP SSE

        or RuntimeConfig attributes can be passed as keyword arguments.
    """
    default_config = dict(
        cors=global_config.ALLOW_CORS,
        property_handler=kwargs.get("property_handler", PropertyHandler),
        action_handler=kwargs.get("action_handler", ActionHandler),
        event_handler=kwargs.get("event_handler", EventHandler),
        thing_description_handler=kwargs.get("thing_description_handler", ThingDescriptionHandler),
        RW_multiple_properties_handler=kwargs.get("RW_multiple_properties_handler", RWMultiplePropertiesHandler),
        liveness_probe_handler=kwargs.get("liveness_handler", LivenessProbeHandler),
        readiness_probe_handler=kwargs.get("readiness_handler", ReadinessProbeHandler),
        stop_handler=kwargs.get("stop_handler", StopHandler),
        thing_description_service=kwargs.get("thing_description_service", ThingDescriptionService),
        thing_repository=kwargs.get("thing_repository", dict()),
        allowed_clients=allowed_clients,
        security_schemes=security_schemes,
    )
    default_config.update(config or dict())
    config = RuntimeConfig(**default_config)
    # need to be extended when more options are added
    super().__init__(
        port=port,
        address=address,
        logger=logger,
        ssl_context=ssl_context,
        config=config,
    )

    self._IP = f"{self.address}:{self.port}"  # TODO, remove this variable later?
    self.id = self._IP
    if self.logger is None:
        self.logger = structlog.get_logger().bind(component="http-server", host=f"{self.address}:{self.port}")

    self.tornado_instance = None
    self.app = Application(
        handlers=[
            (
                r"/stop",
                self.config.stop_handler,
                dict(config=self.config, logger=self.logger, owner_inst=self),
            ),
            (
                r"/liveness",
                self.config.liveness_probe_handler,
                dict(config=self.config, logger=self.logger, owner_inst=self),
            ),
            (
                r"/readiness",
                self.config.readiness_probe_handler,
                dict(config=self.config, logger=self.logger, owner_inst=self),
            ),
        ]
    )
    self.router = ApplicationRouter(self.app, self)

    self.zmq_client_pool = MessageMappedZMQClientPool(
        id=self.id,
        server_ids=[],
        client_ids=[],
        handshake=False,
        poll_timeout=100,
    )
    self.add_things(*(things or []))

add_thing 

add_thing(thing: Thing) -> None
Source code in hololinked/hololinked/server/http/server.py 
374
375
376
def add_thing(self, thing: Thing) -> None:
    self.router.add_thing(thing)
    self.things.append(thing)

stop 

stop(attempt_async_stop: bool = True) -> None

Stop the HTTP server - unreliable, use async_stop() if possible. A stop handler at the path /stop with POST method is already implemented that invokes this method for the clients.

Parameters:

Name Type Description Default

attempt_async_stop

bool

if True, attempts to run the async_stop method to close all connections gracefully.

 True
Source code in hololinked/hololinked/server/http/server.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def stop(self, attempt_async_stop: bool = True) -> None:
    """
    Stop the HTTP server - unreliable, use `async_stop()` if possible.
    A stop handler at the path `/stop` with POST method is already implemented that invokes this
    method for the clients.

    Parameters
    ----------
    attempt_async_stop: bool, default `True`
        if `True`, attempts to run the `async_stop` method to close all connections gracefully.
    """
    if attempt_async_stop:
        run_callable_somehow(self.async_stop())
        return
    self.zmq_client_pool.stop_polling()
    if not self.tornado_instance:
        return
    self.tornado_instance.stop()
    run_callable_somehow(self.tornado_instance.close_all_connections())

async_stop async 

async_stop() -> None

Stop the HTTP server. A stop handler at the path /stop with POST method is already implemented that invokes this method for the clients.

Source code in hololinked/hololinked/server/http/server.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
async def async_stop(self) -> None:
    """
    Stop the HTTP server. A stop handler at the path `/stop` with POST method is already implemented
    that invokes this method for the clients.
    """
    self.zmq_client_pool.stop_polling()
    if not self.tornado_instance:
        return
    try:
        self.tornado_instance.stop()
        await self.tornado_instance.close_all_connections()
    except Exception as ex:
        self.logger.error(
            "error while stopping tornado server, use stop() method "
            + f"from hololinked.server and do not reuse the port - {ex}"
        )

add_property 

add_property(URL_path: str, property: Property | PropertyAffordance, http_methods: str | tuple[str, str, str] = ('GET', 'PUT'), handler: BaseHandler | PropertyHandler = PropertyHandler, **kwargs) -> None

Add a property to be accessible by HTTP.

Parameters:

Name Type Description Default

URL_path

str

URL path to access the property

 required

property

Property | PropertyAffordance

Property (object) to be served or its JSON representation

 required

http_methods

str | tuple[str, str, str]

tuple of http methods to be used for read, write and delete. Use None or omit HTTP method for unsupported operations. For example - for readonly property use ('GET', None, None) or ('GET',)

 ('GET', 'PUT')

handler

BaseHandler | PropertyHandler

custom handler for the property, otherwise the default handler will be used

 PropertyHandler

kwargs

additional keyword arguments to be passed to the handler's init

 {}
Source code in hololinked/hololinked/server/http/server.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def add_property(
    self,
    URL_path: str,
    property: Property | PropertyAffordance,
    http_methods: str | tuple[str, str, str] = ("GET", "PUT"),
    handler: BaseHandler | PropertyHandler = PropertyHandler,
    **kwargs,
) -> None:
    """
    Add a property to be accessible by HTTP.

    Parameters
    ----------
    URL_path: str
        URL path to access the property
    property: Property | PropertyAffordance
        Property (object) to be served or its JSON representation
    http_methods: Tuple[str, str, str]
        tuple of http methods to be used for read, write and delete. Use None or omit HTTP method for
        unsupported operations. For example - for readonly property use ('GET', None, None) or ('GET',)
    handler: BaseHandler | PropertyHandler, optional
        custom handler for the property, otherwise the default handler will be used
    kwargs: dict
        additional keyword arguments to be passed to the handler's __init__
    """
    if not isinstance(property, (Property, PropertyAffordance)):
        raise TypeError(f"property should be of type Property, given type {type(property)}")
    if not issubklass(handler, BaseHandler):
        raise TypeError(f"handler should be subclass of BaseHandler, given type {type(handler)}")
    if isinstance(property, Property):
        property = property.to_affordance()
    read_http_method = write_http_method = delete_http_method = None
    http_methods = self.router.adapt_http_methods(http_methods)
    if len(http_methods) == 1:
        read_http_method = http_methods[0]
    elif len(http_methods) == 2:
        read_http_method, write_http_method = http_methods
    elif len(http_methods) == 3:
        read_http_method, write_http_method, delete_http_method = http_methods
    if read_http_method != "GET":
        raise ValueError("read method should be GET")
    if write_http_method and write_http_method not in ["POST", "PUT"]:
        raise ValueError("write method should be POST or PUT")
    if delete_http_method and delete_http_method != "DELETE":
        raise ValueError("delete method should be DELETE")
    kwargs["resource"] = property
    kwargs["logger"] = self.logger
    kwargs["config"] = self.config
    kwargs["metadata"] = HandlerMetadata(http_methods=http_methods)
    self.router.add_rule(affordance=property, URL_path=URL_path, handler=handler, kwargs=kwargs)

add_action 

add_action(URL_path: str, action: Action | ActionAffordance, http_method: str | None = 'POST', handler: BaseHandler | ActionHandler = ActionHandler, **kwargs) -> None

Add an action to be accessible by HTTP

Parameters:

Name Type Description Default

URL_path

str

URL path to access the action

 required

action

Action | ActionAffordance

Action (object) to be served or its JSON representation

 required

http_method

str | None

http method to be used for the action

 'POST'

handler

BaseHandler | ActionHandler

custom handler for the action

 ActionHandler

kwargs

additional keyword arguments to be passed to the handler's init

 {}
Source code in hololinked/hololinked/server/http/server.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def add_action(
    self,
    URL_path: str,
    action: Action | ActionAffordance,
    http_method: str | None = "POST",
    handler: BaseHandler | ActionHandler = ActionHandler,
    **kwargs,
) -> None:
    """
    Add an action to be accessible by HTTP

    Parameters
    ----------
    URL_path: str
        URL path to access the action
    action: Action | ActionAffordance
        Action (object) to be served or its JSON representation
    http_method: str
        http method to be used for the action
    handler: BaseHandler | ActionHandler, optional
        custom handler for the action
    kwargs: dict
        additional keyword arguments to be passed to the handler's __init__
    """
    if not isinstance(action, (Action, ActionAffordance)):
        raise TypeError(f"Given action should be of type Action or ActionAffordance, given type {type(action)}")
    if not issubklass(handler, BaseHandler):
        raise TypeError(f"handler should be subclass of BaseHandler, given type {type(handler)}")
    http_methods = self.router.adapt_http_methods(http_method)
    if isinstance(action, Action):
        action = action.to_affordance()  # type: ActionAffordance
    kwargs["resource"] = action
    kwargs["config"] = self.config
    kwargs["logger"] = self.logger
    kwargs["metadata"] = HandlerMetadata(http_methods=http_methods)
    self.router.add_rule(affordance=action, URL_path=URL_path, handler=handler, kwargs=kwargs)

add_event 

add_event(URL_path: str, event: Event | EventAffordance | PropertyAffordance, handler: BaseHandler | EventHandler = EventHandler, **kwargs) -> None

Add an event to be accessible by HTTP server; only GET method is supported for events.

Parameters:

Name Type Description Default

URL_path

str

URL path to access the event

 required

event

Event | EventAffordance | PropertyAffordance

Event (object) to be served or its JSON representation

 required

handler

BaseHandler | EventHandler

custom handler for the event

 EventHandler

kwargs

additional keyword arguments to be passed to the handler's init

 {}
Source code in hololinked/hololinked/server/http/server.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def add_event(
    self,
    URL_path: str,
    event: Event | EventAffordance | PropertyAffordance,
    handler: BaseHandler | EventHandler = EventHandler,
    **kwargs,
) -> None:
    """
    Add an event to be accessible by HTTP server; only GET method is supported for events.

    Parameters
    ----------
    URL_path: str
        URL path to access the event
    event: Event | EventAffordance
        Event (object) to be served or its JSON representation
    handler: BaseHandler | EventHandler, optional
        custom handler for the event
    kwargs: dict
        additional keyword arguments to be passed to the handler's __init__
    """
    if not isinstance(event, (Event, EventAffordance)) and (
        not isinstance(event, PropertyAffordance) or not event.observable
    ):
        raise TypeError(f"event should be of type Event or EventAffordance, given type {type(event)}")
    if not issubklass(handler, BaseHandler):
        raise TypeError(f"handler should be subclass of BaseHandler, given type {type(handler)}")
    if isinstance(event, Event):
        event = event.to_affordance()
    kwargs["resource"] = event
    kwargs["config"] = self.config
    kwargs["logger"] = self.logger
    kwargs["metadata"] = HandlerMetadata(http_methods=("GET",))
    self.router.add_rule(affordance=event, URL_path=URL_path, handler=handler, kwargs=kwargs)