BaseHandler

hololinked.server.http.handlers.BaseHandler

Bases: RequestHandler

Base request handler for running operations on the Thing.

Source code in repo/hololinked/hololinked/server/http/handlers.py

class BaseHandler(RequestHandler):
    """Base request handler for running operations on the `Thing`."""

    # Would be a Controller in layered architecture.

    def initialize(  # ty: ignore[invalid-method-override]
        self,
        resource: InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance,
        config: RuntimeConfig,
        logger: structlog.stdlib.BoundLogger,
        thing: Thing,
        metadata: HandlerMetadata | None = None,
    ) -> None:
        """
        Set up the handler with the affordance it serves and the server's runtime configuration.

        Parameters
        ----------
        resource: InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance
            dataclass representation of `Thing`'s exposed object that can quickly convert to a ZMQ Request object
        metadata: HandlerMetadata | None,
            additional metadata about the resource, like allowed HTTP methods
        thing: Thing
            the `Thing` this handler serves
        """
        from hololinked.server.http.config import HandlerMetadata

        self.resource = resource  # type: InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance
        self.config = config
        self.logger = logger.bind(
            resource=resource.name,
            what=resource.what,
            thing_id=resource.thing_id,
            path=self.request.path,
            impl=self.__class__.__name__,
        )
        self.thing: Thing = thing
        self.thing_id = self.resource.thing_id
        self.allowed_clients = self.config.allowed_clients
        self.security_schemes = self.config.security_schemes
        self.metadata = metadata or HandlerMetadata()
        self.userinfo: dict[str, Any] | None = None

    async def has_access_control(self) -> bool:
        """
        Checks if a client is an allowed client and enforces security schemes.

        Custom web request handlers can use this property to check if a client has access control on the server or `Thing`
        and let this property automatically generate a 401/403.

        Returns
        -------
        bool
            `True` if the client may proceed, `False` if a 401/403 was already set on the response
        """
        if not self.allowed_clients and not self.security_schemes:
            return True
        # First check if the client is allowed to access the server
        origin = self.request.headers.get("Origin")
        if (
            self.allowed_clients
            and origin is not None
            and (origin not in self.allowed_clients and origin + "/" not in self.allowed_clients)
        ):
            self.set_status(401, "Unauthorized")
            return False
        # Then check an authentication scheme either if the client is allowed
        # or if there is no such list of allowed clients
        if not self.security_schemes:
            self.logger.debug("no security schemes defined, allowing access")
            return True
        if await self.is_authenticated():
            self.logger.info("client authenticated successfully")
            if not self.userinfo:
                return True
            if await self.is_authorized():
                return True
            self.set_status(403, "Forbidden")
            self.logger.info("insufficient permissions")
            return False
        self.set_status(401, "Unauthorized")
        self.logger.info("client authentication failed or is not authorized to proceed")
        return False  # keep False always at the end

    async def is_authenticated(self) -> bool:
        """
        Enforces authentication using the defined security schemes, freshly computed everytime.

        Returns
        -------
        bool
            `True` if one of the security schemes accepted the credentials supplied with the request
        """
        authenticated = False
        # 1. Basic Authentication
        authorization_header = self.request.headers.get("Authorization", None)  # type: str
        if authorization_header and "basic " in authorization_header[:10].lower():  # basic <base64-encoded>
            for security_scheme in self.security_schemes or []:
                if isinstance(security_scheme, (BcryptBasicSecurity, Argon2BasicSecurity)):
                    try:
                        self.logger.info(
                            "authenticating client",
                            origin=self.request.headers.get("Origin"),
                            security_scheme=security_scheme.__class__.__name__,
                        )
                        if security_scheme.expect_base64:  # ty: ignore[unresolved-attribute]
                            authenticated = security_scheme.validate_base64(  # ty: ignore[unresolved-attribute]
                                authorization_header.split()[1]
                            )
                        else:
                            authenticated = security_scheme.validate_input(  # ty: ignore[unresolved-attribute]
                                username=authorization_header.split()[1].split(":", 1)[0],
                                password=authorization_header.split()[1].split(":", 1)[1],
                            )
                    except Exception as ex:
                        self.logger.error(f"error while authenticating client - {str(ex)}")
                    if authenticated:
                        return True
        # 2. API Key Authentication
        apikey = self.request.headers.get("X-API-Key", None)  # type: str
        if apikey:
            for security_scheme in self.security_schemes or []:
                if isinstance(security_scheme, APIKeySecurity):
                    try:
                        self.logger.info(
                            "authenticating client with API key",
                            origin=self.request.headers.get("Origin"),
                            security_scheme=security_scheme.__class__.__name__,
                        )
                        authenticated = security_scheme.validate_input(apikey)  # ty: ignore[unresolved-attribute]
                    except Exception as ex:
                        self.logger.error(f"error while authenticating client with API key - {str(ex)}")
                    if authenticated:
                        return True
        # 3. JWT from OIDC
        if authorization_header and "bearer " in authorization_header[:10].lower():  # bearer <bla-bla>
            for security_scheme in self.security_schemes or []:
                if isinstance(security_scheme, OIDCSecurity):
                    try:
                        self.logger.info(
                            "authenticating client with OIDC JWT",
                            origin=self.request.headers.get("Origin"),
                            security_scheme=security_scheme.__class__.__name__,
                        )
                        jwt = authorization_header.split(maxsplit=1)[1]
                        self.userinfo = security_scheme.userinfo(jwt)  # ty: ignore[unresolved-attribute]
                        if not self.userinfo:
                            continue
                        authenticated = True
                    except Exception as ex:
                        self.logger.error(f"error while authenticating client with OIDC JWT - {str(ex)}")
                    if authenticated:
                        return True
        return authenticated

    async def is_authorized(self) -> bool:
        """
        Enforces authorization using the defined security schemes, freshly computed everytime.

        Do not call this method without doing authentication first and having a userinfo property.

        Returns
        -------
        bool
            `True` if there is no user info to check against, or the user carries one of the allowed roles
        """
        if not self.userinfo:
            return True
        for security_scheme in self.security_schemes or []:
            if isinstance(security_scheme, OIDCSecurity):
                return security_scheme.user_has_role(self.userinfo)  # ty: ignore[unresolved-attribute]
        return False

    def set_access_control_allow_headers(self) -> None:
        """
        For credential login, access control allow headers cannot be a wildcard '*'.

        Some requests require exact list of allowed headers for the client to access the response.
        """
        headers = ", ".join(self.request.headers.keys())
        if self.request.headers.get("Access-Control-Request-Headers", None):
            headers += ", " + self.request.headers["Access-Control-Request-Headers"]
        self.set_header("Access-Control-Allow-Headers", headers)

    def set_custom_default_headers(self) -> None:
        """
        Sets general default headers, override in child classes to add more headers.

        ```yaml
        Content-Type: application/json
        Access-Control-Allow-Origin: <client>
        ```
        """
        # Access-Control-Allow-Credentials: true # only for cookie auth
        if self.config.cors:
            # For credential login, access control allow origin cannot be '*',
            # See: https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS#examples_of_access_control_scenarios
            self.set_header("Access-Control-Allow-Origin", "*")

    def get_execution_parameters(
        self,
    ) -> tuple[
        SchedulerExecutionContext,
        ThingExecutionContext,
        LocalExecutionContext,
        SerializableData,
    ]:
        """
        Aggregates all arguments to a standard dataclasses from the query parameters.

        Retrieves execution context (like oneway calls, fetching executing
        logs), timeouts, etc. Non recognized arguments are passed as additional payload to the `Thing`.

        An example would be the following URL:

        ```
        http://localhost:8080/property/temperature?oneway=true&invokationTimeout=5&some_arg=42
        ```

        scheduler execution context would have `oneway` set to true & `invokationTimeout` set to 5 seconds,
        local execution context would be empty as no such arguments were passed,
        and additional payload would have `{"some_arg": 42}` as its value.

        Returns
        -------
        tuple[
            SchedulerExecutionContext,
            ThingExecutionContext,
            LocalExecutionContext,
            SerializableData,
        ]
            scheduler execution context, thing execution context, local execution context and payload (if any)
        """
        arguments = dict()
        if len(self.request.query_arguments) == 0:
            return (
                default_scheduler_execution_context,
                default_thing_execution_context,
                LocalExecutionContext(),
                SerializableNone,
            )
        for key, value in self.request.query_arguments.items():
            if key == "messageID":
                # not a JSON value, a hex ID like `1765e270` becomes a float
                arguments[key] = value[0].decode("utf-8")
            elif len(value) == 1:
                try:
                    arguments[key] = Serializers.json.loads(value[0])
                except MsgspecJSONDecodeError:
                    arguments[key] = value[0].decode("utf-8")
            else:
                final_value = []
                for val in value:
                    try:
                        final_value.append(Serializers.json.loads(val))
                    except MsgspecJSONDecodeError:
                        final_value.append(val.decode("utf-8"))
                arguments[key] = final_value
        # if self.resource.request_as_argument:
        #     arguments['request'] = self.request # find some way to pass the request object to the thing
        thing_execution_context = ThingExecutionContext(
            fetch_execution_logs=bool(arguments.pop("fetchExecutionLogs", False))
        )
        scheduler_execution_context = SchedulerExecutionContext(
            invokation_timeout=arguments.pop(
                "invokationTimeout", default_scheduler_execution_context.invokation_timeout
            ),
            execution_timeout=arguments.pop("executionTimeout", default_scheduler_execution_context.execution_timeout),
            oneway=arguments.pop("oneway", default_scheduler_execution_context.oneway),
        )
        local_execution_context = LocalExecutionContext(
            noblock=arguments.pop("noblock", None),
            messageID=arguments.pop("messageID", None),
        )
        additional_payload = SerializableNone if not arguments else SerializableData(arguments)  # application/json
        return scheduler_execution_context, thing_execution_context, local_execution_context, additional_payload

    @property
    def message_id(self) -> str | None:
        """Retrieves the message id from the request headers."""
        try:
            return self._message_id
        except AttributeError:
            message_id = self.request.headers.get("X-Message-ID", None)
            if not message_id:
                _, _, local_execution_context, _ = self.get_execution_parameters()
                # TODO avoid calling get_execution_parameters twice in the same request
                message_id = local_execution_context.messageID
            self._message_id = message_id
            return message_id

    def get_request_payload(self) -> tuple[SerializableData, PreserializedData]:
        """
        Retrieves the payload from the request body, does not necessarily deserialize it.

        Returns
        -------
        tuple[SerializableData, PreserializedData]
            the request body as serializable data and as preserialized data - the body is carried by whichever
            of the two matches the request's content type

        Raises
        ------
        ValueError
            if the request declares a content type that is not supported
        """
        payload = SerializableData(value=None)
        preserialized_payload = PreserializedData(value=b"")
        if self.request.body:
            if self.request.headers.get("Content-Type", "application/json") in Serializers.allowed_content_types:
                payload.value = self.request.body
                payload.content_type = self.request.headers.get("Content-Type", "application/json")
            elif global_config.ALLOW_UNKNOWN_SERIALIZATION:
                preserialized_payload.value = self.request.body
                content_type = self.request.headers.get("Content-Type", None)
                if content_type:
                    preserialized_payload.content_type = content_type
            else:
                raise ValueError("Content-Type not supported")
                # NOTE that was assume that the content type is JSON even if unspecified in the header.
                # This error will be raised only when a specified content type is not supported.
        return payload, preserialized_payload

    async def get(self) -> None:  # ty: ignore[invalid-method-override]
        """Runs property or action if accessible by 'GET' method. Default for property reads."""
        raise NotImplementedError("implement GET request method in child handler class")

    async def post(self) -> None:  # ty: ignore[invalid-method-override]
        """Runs property or action if accessible by 'POST' method. Default for action execution."""
        raise NotImplementedError("implement POST request method in child handler class")

    async def put(self) -> None:  # ty: ignore[invalid-method-override]
        """Runs property or action if accessible by 'PUT' method. Default for property writes."""
        raise NotImplementedError("implement PUT request method in child handler class")

    async def delete(self) -> None:  # ty: ignore[invalid-method-override]
        """
        Runs property or action if accessible by 'DELETE' method.

        Default for property deletes (not a valid operation as per web of things semantics).
        """
        raise NotImplementedError("implement DELETE request method in child handler class")

    async def is_method_allowed(self, method: str) -> bool:
        """Checks if the method is allowed for the property."""
        raise NotImplementedError("implement is_method_allowed in child handler class")

Attributes

message_id property

message_id: str | None

Retrieves the message id from the request headers.

Functions

initialize

initialize(resource: InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance, config: RuntimeConfig, logger: BoundLogger, thing: Thing, metadata: HandlerMetadata | None = None) -> None

Set up the handler with the affordance it serves and the server's runtime configuration.

Parameters:

Name	Type	Description	Default
resource	InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance	dataclass representation of Thing's exposed object that can quickly convert to a ZMQ Request object	required
metadata	HandlerMetadata | None	additional metadata about the resource, like allowed HTTP methods	None
thing	Thing	the Thing this handler serves	required

Source code in repo/hololinked/hololinked/server/http/handlers.py

def initialize(  # ty: ignore[invalid-method-override]
    self,
    resource: InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance,
    config: RuntimeConfig,
    logger: structlog.stdlib.BoundLogger,
    thing: Thing,
    metadata: HandlerMetadata | None = None,
) -> None:
    """
    Set up the handler with the affordance it serves and the server's runtime configuration.

    Parameters
    ----------
    resource: InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance
        dataclass representation of `Thing`'s exposed object that can quickly convert to a ZMQ Request object
    metadata: HandlerMetadata | None,
        additional metadata about the resource, like allowed HTTP methods
    thing: Thing
        the `Thing` this handler serves
    """
    from hololinked.server.http.config import HandlerMetadata

    self.resource = resource  # type: InteractionAffordance | PropertyAffordance | ActionAffordance | EventAffordance
    self.config = config
    self.logger = logger.bind(
        resource=resource.name,
        what=resource.what,
        thing_id=resource.thing_id,
        path=self.request.path,
        impl=self.__class__.__name__,
    )
    self.thing: Thing = thing
    self.thing_id = self.resource.thing_id
    self.allowed_clients = self.config.allowed_clients
    self.security_schemes = self.config.security_schemes
    self.metadata = metadata or HandlerMetadata()
    self.userinfo: dict[str, Any] | None = None

has_access_control async

has_access_control() -> bool

Checks if a client is an allowed client and enforces security schemes.

Custom web request handlers can use this property to check if a client has access control on the server or Thing and let this property automatically generate a 401/403.

Returns:

Type	Description
bool	True if the client may proceed, False if a 401/403 was already set on the response

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def has_access_control(self) -> bool:
    """
    Checks if a client is an allowed client and enforces security schemes.

    Custom web request handlers can use this property to check if a client has access control on the server or `Thing`
    and let this property automatically generate a 401/403.

    Returns
    -------
    bool
        `True` if the client may proceed, `False` if a 401/403 was already set on the response
    """
    if not self.allowed_clients and not self.security_schemes:
        return True
    # First check if the client is allowed to access the server
    origin = self.request.headers.get("Origin")
    if (
        self.allowed_clients
        and origin is not None
        and (origin not in self.allowed_clients and origin + "/" not in self.allowed_clients)
    ):
        self.set_status(401, "Unauthorized")
        return False
    # Then check an authentication scheme either if the client is allowed
    # or if there is no such list of allowed clients
    if not self.security_schemes:
        self.logger.debug("no security schemes defined, allowing access")
        return True
    if await self.is_authenticated():
        self.logger.info("client authenticated successfully")
        if not self.userinfo:
            return True
        if await self.is_authorized():
            return True
        self.set_status(403, "Forbidden")
        self.logger.info("insufficient permissions")
        return False
    self.set_status(401, "Unauthorized")
    self.logger.info("client authentication failed or is not authorized to proceed")
    return False  # keep False always at the end

is_authenticated async

is_authenticated() -> bool

Enforces authentication using the defined security schemes, freshly computed everytime.

Returns:

Type	Description
bool	True if one of the security schemes accepted the credentials supplied with the request

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def is_authenticated(self) -> bool:
    """
    Enforces authentication using the defined security schemes, freshly computed everytime.

    Returns
    -------
    bool
        `True` if one of the security schemes accepted the credentials supplied with the request
    """
    authenticated = False
    # 1. Basic Authentication
    authorization_header = self.request.headers.get("Authorization", None)  # type: str
    if authorization_header and "basic " in authorization_header[:10].lower():  # basic <base64-encoded>
        for security_scheme in self.security_schemes or []:
            if isinstance(security_scheme, (BcryptBasicSecurity, Argon2BasicSecurity)):
                try:
                    self.logger.info(
                        "authenticating client",
                        origin=self.request.headers.get("Origin"),
                        security_scheme=security_scheme.__class__.__name__,
                    )
                    if security_scheme.expect_base64:  # ty: ignore[unresolved-attribute]
                        authenticated = security_scheme.validate_base64(  # ty: ignore[unresolved-attribute]
                            authorization_header.split()[1]
                        )
                    else:
                        authenticated = security_scheme.validate_input(  # ty: ignore[unresolved-attribute]
                            username=authorization_header.split()[1].split(":", 1)[0],
                            password=authorization_header.split()[1].split(":", 1)[1],
                        )
                except Exception as ex:
                    self.logger.error(f"error while authenticating client - {str(ex)}")
                if authenticated:
                    return True
    # 2. API Key Authentication
    apikey = self.request.headers.get("X-API-Key", None)  # type: str
    if apikey:
        for security_scheme in self.security_schemes or []:
            if isinstance(security_scheme, APIKeySecurity):
                try:
                    self.logger.info(
                        "authenticating client with API key",
                        origin=self.request.headers.get("Origin"),
                        security_scheme=security_scheme.__class__.__name__,
                    )
                    authenticated = security_scheme.validate_input(apikey)  # ty: ignore[unresolved-attribute]
                except Exception as ex:
                    self.logger.error(f"error while authenticating client with API key - {str(ex)}")
                if authenticated:
                    return True
    # 3. JWT from OIDC
    if authorization_header and "bearer " in authorization_header[:10].lower():  # bearer <bla-bla>
        for security_scheme in self.security_schemes or []:
            if isinstance(security_scheme, OIDCSecurity):
                try:
                    self.logger.info(
                        "authenticating client with OIDC JWT",
                        origin=self.request.headers.get("Origin"),
                        security_scheme=security_scheme.__class__.__name__,
                    )
                    jwt = authorization_header.split(maxsplit=1)[1]
                    self.userinfo = security_scheme.userinfo(jwt)  # ty: ignore[unresolved-attribute]
                    if not self.userinfo:
                        continue
                    authenticated = True
                except Exception as ex:
                    self.logger.error(f"error while authenticating client with OIDC JWT - {str(ex)}")
                if authenticated:
                    return True
    return authenticated

is_authorized async

is_authorized() -> bool

Enforces authorization using the defined security schemes, freshly computed everytime.

Do not call this method without doing authentication first and having a userinfo property.

Returns:

Type	Description
bool	True if there is no user info to check against, or the user carries one of the allowed roles

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def is_authorized(self) -> bool:
    """
    Enforces authorization using the defined security schemes, freshly computed everytime.

    Do not call this method without doing authentication first and having a userinfo property.

    Returns
    -------
    bool
        `True` if there is no user info to check against, or the user carries one of the allowed roles
    """
    if not self.userinfo:
        return True
    for security_scheme in self.security_schemes or []:
        if isinstance(security_scheme, OIDCSecurity):
            return security_scheme.user_has_role(self.userinfo)  # ty: ignore[unresolved-attribute]
    return False

is_method_allowed async

is_method_allowed(method: str) -> bool

Checks if the method is allowed for the property.

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def is_method_allowed(self, method: str) -> bool:
    """Checks if the method is allowed for the property."""
    raise NotImplementedError("implement is_method_allowed in child handler class")

set_access_control_allow_headers

set_access_control_allow_headers() -> None

For credential login, access control allow headers cannot be a wildcard '*'.

Some requests require exact list of allowed headers for the client to access the response.

Source code in repo/hololinked/hololinked/server/http/handlers.py

def set_access_control_allow_headers(self) -> None:
    """
    For credential login, access control allow headers cannot be a wildcard '*'.

    Some requests require exact list of allowed headers for the client to access the response.
    """
    headers = ", ".join(self.request.headers.keys())
    if self.request.headers.get("Access-Control-Request-Headers", None):
        headers += ", " + self.request.headers["Access-Control-Request-Headers"]
    self.set_header("Access-Control-Allow-Headers", headers)

set_custom_default_headers

set_custom_default_headers() -> None

Sets general default headers, override in child classes to add more headers.

Content-Type: application/json
Access-Control-Allow-Origin: <client>

Source code in repo/hololinked/hololinked/server/http/handlers.py

def set_custom_default_headers(self) -> None:
    """
    Sets general default headers, override in child classes to add more headers.

    ```yaml
    Content-Type: application/json
    Access-Control-Allow-Origin: <client>
    ```
    """
    # Access-Control-Allow-Credentials: true # only for cookie auth
    if self.config.cors:
        # For credential login, access control allow origin cannot be '*',
        # See: https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS#examples_of_access_control_scenarios
        self.set_header("Access-Control-Allow-Origin", "*")

get_execution_parameters

get_execution_parameters() -> tuple[SchedulerExecutionContext, ThingExecutionContext, LocalExecutionContext, SerializableData]

Aggregates all arguments to a standard dataclasses from the query parameters.

Retrieves execution context (like oneway calls, fetching executing logs), timeouts, etc. Non recognized arguments are passed as additional payload to the Thing.

An example would be the following URL:

http://localhost:8080/property/temperature?oneway=true&invokationTimeout=5&some_arg=42

scheduler execution context would have oneway set to true & invokationTimeout set to 5 seconds, local execution context would be empty as no such arguments were passed, and additional payload would have {"some_arg": 42} as its value.

Returns:

Type	Description
tuple[	SchedulerExecutionContext, ThingExecutionContext, LocalExecutionContext, SerializableData,
]	scheduler execution context, thing execution context, local execution context and payload (if any)

Source code in repo/hololinked/hololinked/server/http/handlers.py

def get_execution_parameters(
    self,
) -> tuple[
    SchedulerExecutionContext,
    ThingExecutionContext,
    LocalExecutionContext,
    SerializableData,
]:
    """
    Aggregates all arguments to a standard dataclasses from the query parameters.

    Retrieves execution context (like oneway calls, fetching executing
    logs), timeouts, etc. Non recognized arguments are passed as additional payload to the `Thing`.

    An example would be the following URL:

    ```
    http://localhost:8080/property/temperature?oneway=true&invokationTimeout=5&some_arg=42
    ```

    scheduler execution context would have `oneway` set to true & `invokationTimeout` set to 5 seconds,
    local execution context would be empty as no such arguments were passed,
    and additional payload would have `{"some_arg": 42}` as its value.

    Returns
    -------
    tuple[
        SchedulerExecutionContext,
        ThingExecutionContext,
        LocalExecutionContext,
        SerializableData,
    ]
        scheduler execution context, thing execution context, local execution context and payload (if any)
    """
    arguments = dict()
    if len(self.request.query_arguments) == 0:
        return (
            default_scheduler_execution_context,
            default_thing_execution_context,
            LocalExecutionContext(),
            SerializableNone,
        )
    for key, value in self.request.query_arguments.items():
        if key == "messageID":
            # not a JSON value, a hex ID like `1765e270` becomes a float
            arguments[key] = value[0].decode("utf-8")
        elif len(value) == 1:
            try:
                arguments[key] = Serializers.json.loads(value[0])
            except MsgspecJSONDecodeError:
                arguments[key] = value[0].decode("utf-8")
        else:
            final_value = []
            for val in value:
                try:
                    final_value.append(Serializers.json.loads(val))
                except MsgspecJSONDecodeError:
                    final_value.append(val.decode("utf-8"))
            arguments[key] = final_value
    # if self.resource.request_as_argument:
    #     arguments['request'] = self.request # find some way to pass the request object to the thing
    thing_execution_context = ThingExecutionContext(
        fetch_execution_logs=bool(arguments.pop("fetchExecutionLogs", False))
    )
    scheduler_execution_context = SchedulerExecutionContext(
        invokation_timeout=arguments.pop(
            "invokationTimeout", default_scheduler_execution_context.invokation_timeout
        ),
        execution_timeout=arguments.pop("executionTimeout", default_scheduler_execution_context.execution_timeout),
        oneway=arguments.pop("oneway", default_scheduler_execution_context.oneway),
    )
    local_execution_context = LocalExecutionContext(
        noblock=arguments.pop("noblock", None),
        messageID=arguments.pop("messageID", None),
    )
    additional_payload = SerializableNone if not arguments else SerializableData(arguments)  # application/json
    return scheduler_execution_context, thing_execution_context, local_execution_context, additional_payload

get_request_payload

get_request_payload() -> tuple[SerializableData, PreserializedData]

Retrieves the payload from the request body, does not necessarily deserialize it.

Returns:

Type	Description
tuple[SerializableData, PreserializedData]	the request body as serializable data and as preserialized data - the body is carried by whichever of the two matches the request's content type

Raises:

Type	Description
ValueError	if the request declares a content type that is not supported

Source code in repo/hololinked/hololinked/server/http/handlers.py

def get_request_payload(self) -> tuple[SerializableData, PreserializedData]:
    """
    Retrieves the payload from the request body, does not necessarily deserialize it.

    Returns
    -------
    tuple[SerializableData, PreserializedData]
        the request body as serializable data and as preserialized data - the body is carried by whichever
        of the two matches the request's content type

    Raises
    ------
    ValueError
        if the request declares a content type that is not supported
    """
    payload = SerializableData(value=None)
    preserialized_payload = PreserializedData(value=b"")
    if self.request.body:
        if self.request.headers.get("Content-Type", "application/json") in Serializers.allowed_content_types:
            payload.value = self.request.body
            payload.content_type = self.request.headers.get("Content-Type", "application/json")
        elif global_config.ALLOW_UNKNOWN_SERIALIZATION:
            preserialized_payload.value = self.request.body
            content_type = self.request.headers.get("Content-Type", None)
            if content_type:
                preserialized_payload.content_type = content_type
        else:
            raise ValueError("Content-Type not supported")
            # NOTE that was assume that the content type is JSON even if unspecified in the header.
            # This error will be raised only when a specified content type is not supported.
    return payload, preserialized_payload

get async

get() -> None

Runs property or action if accessible by 'GET' method. Default for property reads.

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def get(self) -> None:  # ty: ignore[invalid-method-override]
    """Runs property or action if accessible by 'GET' method. Default for property reads."""
    raise NotImplementedError("implement GET request method in child handler class")

post async

post() -> None

Runs property or action if accessible by 'POST' method. Default for action execution.

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def post(self) -> None:  # ty: ignore[invalid-method-override]
    """Runs property or action if accessible by 'POST' method. Default for action execution."""
    raise NotImplementedError("implement POST request method in child handler class")

put async

put() -> None

Runs property or action if accessible by 'PUT' method. Default for property writes.

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def put(self) -> None:  # ty: ignore[invalid-method-override]
    """Runs property or action if accessible by 'PUT' method. Default for property writes."""
    raise NotImplementedError("implement PUT request method in child handler class")

delete async

delete() -> None

Runs property or action if accessible by 'DELETE' method.

Default for property deletes (not a valid operation as per web of things semantics).

Source code in repo/hololinked/hololinked/server/http/handlers.py

async def delete(self) -> None:  # ty: ignore[invalid-method-override]
    """
    Runs property or action if accessible by 'DELETE' method.

    Default for property deletes (not a valid operation as per web of things semantics).
    """
    raise NotImplementedError("implement DELETE request method in child handler class")