Payloads

The two containers an Operation and a Reply carry their data in.

hololinked.core.eventloop.payloads.SerializableData dataclass

A container for data that can be serialized.

Either provide a serializer or a content type to pick a suitable already supported serializer.

Source code in repo/hololinked/hololinked/core/eventloop/payloads.py

@dataclass
class SerializableData:
    """
    A container for data that can be serialized.

    Either provide a serializer or a content type to pick a suitable already supported serializer.
    """

    value: Any
    serializer: BaseSerializer | None = None
    content_type: str = "application/json"
    _serialized: bytes | None = None

    def serialize(self) -> bytes:
        """
        Serialize the value.

        Returns
        -------
        bytes
            serialized value

        Raises
        ------
        ValueError
            If no suitable serializer is found for the content type.
        """
        if self._serialized is not None:
            return self._serialized
        if isinstance(self.value, byte_types):
            return self.value
        if self.serializer is not None:
            return self.serializer.dumps(self.value)
        serializer = Serializers.content_types.get(self.content_type, None)
        if serializer is not None:
            return serializer.dumps(self.value)
        raise ValueError(f"content type {self.content_type} not supported for serialization")

    def deserialize(self) -> Any:
        """
        Deserialize the value.

        Returns
        -------
        Any
            deserialized value

        Raises
        ------
        ValueError
            If no suitable serializer is found for the content type.
        """
        if not isinstance(self.value, byte_types):
            return self.value
        if self.serializer is not None:
            return self.serializer.loads(self.value)
        serializer = Serializers.content_types.get(self.content_type, None)
        if serializer is not None:
            return serializer.loads(self.value)
        raise ValueError(f"content type {self.content_type} not supported for deserialization")

    def require_serialized(self) -> None:
        """Ensure the value is serialized. Can raise `ValueError` if no suitable serializer is found."""
        self._serialized = self.serialize()

Attributes

value instance-attribute

value: Any

content_type class-attribute instance-attribute

content_type: str = 'application/json'

serializer class-attribute instance-attribute

serializer: BaseSerializer | None = None

Functions

__init__

__init__(value: Any, serializer: BaseSerializer | None = None, content_type: str = 'application/json', _serialized: bytes | None = None) -> None

serialize

serialize() -> bytes

Serialize the value.

Returns:

Type	Description
bytes	serialized value

Raises:

Type	Description
ValueError	If no suitable serializer is found for the content type.

Source code in repo/hololinked/hololinked/core/eventloop/payloads.py

def serialize(self) -> bytes:
    """
    Serialize the value.

    Returns
    -------
    bytes
        serialized value

    Raises
    ------
    ValueError
        If no suitable serializer is found for the content type.
    """
    if self._serialized is not None:
        return self._serialized
    if isinstance(self.value, byte_types):
        return self.value
    if self.serializer is not None:
        return self.serializer.dumps(self.value)
    serializer = Serializers.content_types.get(self.content_type, None)
    if serializer is not None:
        return serializer.dumps(self.value)
    raise ValueError(f"content type {self.content_type} not supported for serialization")

deserialize

deserialize() -> Any

Deserialize the value.

Returns:

Type	Description
Any	deserialized value

Raises:

Type	Description
ValueError	If no suitable serializer is found for the content type.

Source code in repo/hololinked/hololinked/core/eventloop/payloads.py

def deserialize(self) -> Any:
    """
    Deserialize the value.

    Returns
    -------
    Any
        deserialized value

    Raises
    ------
    ValueError
        If no suitable serializer is found for the content type.
    """
    if not isinstance(self.value, byte_types):
        return self.value
    if self.serializer is not None:
        return self.serializer.loads(self.value)
    serializer = Serializers.content_types.get(self.content_type, None)
    if serializer is not None:
        return serializer.loads(self.value)
    raise ValueError(f"content type {self.content_type} not supported for deserialization")

require_serialized

require_serialized() -> None

Ensure the value is serialized. Can raise ValueError if no suitable serializer is found.

Source code in repo/hololinked/hololinked/core/eventloop/payloads.py

def require_serialized(self) -> None:
    """Ensure the value is serialized. Can raise `ValueError` if no suitable serializer is found."""
    self._serialized = self.serialize()

hololinked.core.eventloop.payloads.PreserializedData dataclass

A container for data that is already serialized.

The content type is only a metadata here. The value is expected to be bytes.

Source code in repo/hololinked/hololinked/core/eventloop/payloads.py

@dataclass
class PreserializedData:
    """
    A container for data that is already serialized.

    The content type is only a metadata here. The value is expected to be bytes.
    """

    value: bytes
    content_type: str = "application/octet-stream"

Attributes

value instance-attribute

value: bytes

content_type class-attribute instance-attribute

content_type: str = 'application/octet-stream'

Functions

__init__

__init__(value: bytes, content_type: str = 'application/octet-stream') -> None