hololinked.serializers.serializers.Serializers

A singleton class that holds all serializers and provides a registry for content types. All members are class attributes and settings are applied process-wide (python process).

Registration of serializer is not mandatory for any property, action or event. The default serializer is JSONSerializer, which will be provided to any unregistered object.

Source code in hololinked/hololinked/serializers/serializers.py

class Serializers(metaclass=MappableSingleton):
    """
    A singleton class that holds all serializers and provides a registry for content types.
    All members are class attributes and settings are applied process-wide (python process).

    Registration of serializer is not mandatory for any property, action or event.
    The default serializer is `JSONSerializer`, which will be provided to any unregistered object.
    """

    json = ClassSelector(
        default=JSONSerializer(),
        class_=BaseSerializer,
        class_member=True,
        doc="The default serializer for all properties, actions and events",
    )  # type: BaseSerializer
    """JSON serializer"""

    pickle = ClassSelector(
        default=PickleSerializer(),
        class_=BaseSerializer,
        class_member=True,
        doc="pickle serializer, unsafe without encryption but useful for faster & flexible serialization of python specific types",
    )  # type: BaseSerializer
    """Pickle serializer, use global_config.ALLOW_PICKE to enable it"""

    msgpack = ClassSelector(
        default=MsgpackSerializer(),
        class_=BaseSerializer,
        class_member=True,
        doc="MessagePack serializer, efficient binary format that is both fast & interoperable between languages ",
    )  # type: BaseSerializer
    """MessagePack serializer"""

    text = ClassSelector(
        default=TextSerializer(),
        class_=BaseSerializer,
        class_member=True,
        doc="Text serializer, converts string or string compatible types to bytes and vice versa",
    )  # type: BaseSerializer
    """Text serializer"""

    default = ClassSelector(
        default=json.default,
        class_=BaseSerializer,
        class_member=True,
        doc="The default serialization to be used",
    )  # type: BaseSerializer
    """The default serializer, change value to set a different default serializer"""

    default_content_type = String(
        fget=lambda self: self.default.content_type,
        class_member=True,
        doc="The default content type for the default serializer",
    )  # type: str

    content_types = Parameter(
        default={
            json.default.content_type: json.default,  # as in the default value of the descriptor
            pickle.default.content_type: pickle.default,
            msgpack.default.content_type: msgpack.default,
            text.default.content_type: text.default,
        },
        doc="A dictionary of content types and their serializers",
        readonly=True,
        class_member=True,
    )  # type: dict[str, BaseSerializer]
    """A dictionary of content types and their serializers"""

    allowed_content_types = Parameter(
        default=None,
        class_member=True,
        doc="A list of content types that are usually considered safe and will be supported by default without any configuration",
        readonly=True,
    )  # type: list[str]
    """
    A list of content types that are usually considered safe 
    and will be supported by default without any configuration
    """

    object_content_type_map = Parameter(
        default=dict(),
        class_member=True,
        doc="A dictionary of content types for specific properties, actions and events",
        readonly=True,
    )  # type: dict[str, dict[str, str]]
    """A dictionary of content types for specific properties, actions and events"""

    object_serializer_map = Parameter(
        default=dict(),
        class_member=True,
        doc="A dictionary of serializer for specific properties, actions and events",
        readonly=True,
    )  # type: dict[str, dict[str, BaseSerializer]]
    """A dictionary of serializer for specific properties, actions and events"""

    protocol_serializer_map = Parameter(
        default=dict(),
        class_member=True,
        doc="A dictionary of serializer for a specific protocol",
        readonly=True,
    )  # type: dict[str, BaseSerializer]
    """A dictionary of default serializer for a specific protocol, currently unimplemented"""

    @classmethod
    def register(cls, serializer: BaseSerializer, name: str | None = None, override: bool = False) -> None:
        """
        Register a new serializer. It is recommended to implement a content type property/attribute for the serializer
        to facilitate automatic deserialization on client side, otherwise deserialization is not gauranteed.
        Moreover, the said serializer must be defined on both client and server side if running in a distributed
        environment.

        Parameters
        ----------
        serializer: BaseSerializer
            the serializer to register
        name: str, optional
            the name of the serializer to be accessible under the object namespace. If not provided, the name of the
            serializer class is used.
        override: bool, optional
            whether to override the serializer if the content type is already registered,
            by default False & raises ValueError for duplicate content type. For example, registering
            a custom JSON serializer will conflict with the default JSONSerializer, so set `override=True`.

        Raises
        ------
        ValueError
            if the serializer content type is already registered
        """
        try:
            if serializer.content_type in cls.content_types and not override:
                raise ValueError("content type already registered : {}".format(serializer.content_type))
            cls.content_types[serializer.content_type] = serializer
        except NotImplementedError:
            warnings.warn("serializer does not implement a content type", category=UserWarning)
        cls[name or serializer.__name__] = serializer

    @classmethod
    def for_object(cls, thing_id: str, thing_cls: str, objekt: str) -> BaseSerializer | None:
        """
        Retrieve a serializer for a given property, action or event

        Parameters
        ----------
        thing_id: str | Any
            the id of the Thing or the Thing that owns the property, action or event
        thing_cls: str | Any
            the class name of the Thing or the Thing that owns the property, action or event
        objekt: str
            the name of the property, action or event

        Returns
        -------
        BaseSerializer | JSONSerializer
            the serializer for the property, action or event. If no serializer is found, the default JSONSerializer is
            returned.
        """
        if len(cls.object_serializer_map) == 0 and len(cls.object_content_type_map) == 0:
            return cls.default
        for thing in [thing_id, thing_cls]:  # first thing id, then thing cls
            if thing in cls.object_serializer_map:
                if objekt in cls.object_serializer_map[thing]:
                    return cls.object_serializer_map[thing][objekt]
            if thing in cls.object_content_type_map:
                if objekt in cls.object_content_type_map[thing]:
                    return cls.content_types.get(cls.object_content_type_map[thing][objekt], None)
                    # if said content type has no serializer, return None instead of default serializer
        return cls.default  # JSON is default serializer

    @classmethod
    def get_content_type_for_object(self, thing_id: str, thing_cls: str, objekt: str) -> str:
        """
        Retrieve a content type for a given property, action or event

        Parameters
        ----------
        thing_id: str | Any
            the id of the Thing or the Thing that owns the property, action or event
        thing_cls: str | Any
            the class name of the Thing or the Thing that owns the property, action or event
        objekt: str
            the name of the property, action or event

        Returns
        -------
        str
            the content type for the property, action or event. If no content type is found, the default content type is
            returned.
        """

        if len(self.object_serializer_map) == 0 and len(self.object_content_type_map) == 0:
            return self.default_content_type
        for thing in [thing_id, thing_cls]:  # first thing id, then thing cls
            if thing in self.object_content_type_map:
                if objekt in self.object_content_type_map[thing]:
                    return self.object_content_type_map[thing][objekt]
        return self.default_content_type  # JSON is default serializer

    @classmethod
    def register_for_object(cls, objekt: Any, serializer: BaseSerializer) -> None:
        """
        Register (an existing) serializer for a property, action or event. Other option is to register a content type,
        the effects are similar.

        Parameters
        ----------
        objekt: str | Property | Action | Event
            the property, action or event
        serializer: BaseSerializer
            the serializer to be used
        """
        if not isinstance(serializer, BaseSerializer):
            raise ValueError("serializer must be an instance of BaseSerializer, given : {}".format(type(serializer)))
        from ..core import Action, Event, Property, Thing

        if not isinstance(objekt, (Property, Action, Event)) and not issubklass(objekt, Thing):
            raise ValueError("object must be a Property, Action or Event, or Thing, got : {}".format(type(objekt)))
        if issubklass(objekt, Thing):
            owner = objekt.__name__
        elif not objekt.owner:
            raise ValueError("object owner cannot be determined : {}".format(objekt))
        else:
            owner = objekt.owner.__name__
        if owner not in cls.object_serializer_map:
            cls.object_serializer_map[owner] = dict()
        if issubklass(objekt, Thing):
            cls.object_serializer_map[owner][objekt.__name__] = serializer
        else:
            cls.object_serializer_map[owner][objekt.name] = serializer

    # @validate_call
    @classmethod
    def register_content_type_for_object(cls, objekt: Any, content_type: str) -> None:
        """
        Register content type for a property, action, event, or a `Thing` class to use a specific serializer.
        If no serializer is found, content type could still be used as metadata.

        Parameters
        ----------
        objekt: Property | Action | Event | Thing
            the property, action or event. string is not accepted - use `register_content_type_for_object_by_name()` instead.
        content_type: str
            the content type for the value of the objekt or the serializer to be used

        Raises
        ------
        ValueError
            if the object is not a Property, Action or Event
        """
        from ..core import Action, Event, Property, Thing

        if not isinstance(objekt, (Property, Action, Event)) and not issubklass(objekt, Thing):
            raise ValueError("object must be a Property, Action or Event, got : {}".format(type(objekt)))
        if issubklass(objekt, Thing):
            owner = objekt.__name__
        elif not objekt.owner:
            raise ValueError("object owner cannot be determined, cannot register content type: {}".format(objekt))
        else:
            owner = objekt.owner.__name__
        if owner not in cls.object_content_type_map:
            cls.object_content_type_map[owner] = dict()
        if issubklass(objekt, Thing):
            cls.object_content_type_map[owner][objekt.__name__] = content_type
            # its a redundant key, TODO - may be there is a better way to structure this map
        else:
            cls.object_content_type_map[owner][objekt.name] = content_type

    # @validate_call
    @classmethod
    def register_content_type_for_object_per_thing_instance(
        cls,
        thing_id: str,
        objekt: str | Any,
        content_type: str,
    ) -> None:
        """
        Register a content type for a property, action or event to use a specific serializer. Other option is
        to register a serializer directly, the effects are similar. If no serializer is found,
        content type could still be used as metadata.

        Parameters
        ----------
        thing_id: str
            the id of the Thing that owns the property, action or event
        objekt: str
            the name of the property, action or event
        content_type: str
            the content type to be used
        """
        from ..core import Action, Event, Property

        if not isinstance(objekt, (Property, Action, Event, str)):
            raise ValueError("object must be a Property, Action or Event, got : {}".format(type(objekt)))
        if not isinstance(objekt, str):
            objekt = objekt.name
        if thing_id not in cls.object_content_type_map:
            cls.object_content_type_map[thing_id] = dict()
        cls.object_content_type_map[thing_id][objekt] = content_type

    @classmethod
    def register_content_type_for_thing_instance(cls, thing_id: str, content_type: str) -> None:
        """
        Register a content type for a specific Thing instance.

        Parameters
        ----------
        thing_id: str
            the id of the Thing
        content_type: str
            the content type to be used
        """
        cls.object_content_type_map[thing_id][thing_id] = content_type
        # remember, its a redundant key, TODO

    @classmethod
    def register_for_object_per_thing_instance(cls, thing_id: str, objekt: str, serializer: BaseSerializer) -> None:
        """
        Register a serializer for a property, action or event for a specific Thing instance.
        If no serializer is found, content type could still be used as metadata.

        Parameters
        ----------
        thing_id: str
            the id of the Thing that owns the property, action or event
        objekt: str
            the name of the property, action or event
        serializer: BaseSerializer
            the serializer to be used
        """
        if thing_id not in cls.object_serializer_map:
            cls.object_serializer_map[thing_id] = dict()
        cls.object_serializer_map[thing_id][objekt] = serializer

    @classmethod
    def register_for_thing_instance(cls, thing_id: str, serializer: BaseSerializer) -> None:
        """
        Register a serializer for a specific Thing instance.

        Parameters
        ----------
        thing_id: str
            the id of the Thing
        serializer: BaseSerializer
            the serializer to be used
        """
        if thing_id not in cls.object_serializer_map:
            cls.object_serializer_map[thing_id] = dict()
        cls.object_serializer_map[thing_id][thing_id] = serializer

    @classmethod
    def reset(cls) -> None:
        """Reset the serializer registry"""
        cls.object_content_type_map.clear()
        cls.object_serializer_map.clear()
        cls.protocol_serializer_map.clear()
        cls.default = cls.json

    @allowed_content_types.getter
    def get_allowed_content_types(cls) -> list[str]:
        """Get a list of all allowed content types for serialization"""
        _allowed_content_types = list(cls.content_types.keys())
        _allowed_content_types.remove(cls.pickle.content_type)
        if global_config.ALLOW_PICKLE:
            _allowed_content_types.append(cls.pickle.content_type)
        return _allowed_content_types

Functions

for_object classmethod

for_object(thing_id: str, thing_cls: str, objekt: str) -> BaseSerializer | None

Retrieve a serializer for a given property, action or event

Parameters:

Name	Type	Description	Default
thing_id	str	the id of the Thing or the Thing that owns the property, action or event	required
thing_cls	str	the class name of the Thing or the Thing that owns the property, action or event	required
objekt	str	the name of the property, action or event	required

Returns:

Type	Description
BaseSerializer | JSONSerializer	the serializer for the property, action or event. If no serializer is found, the default JSONSerializer is returned.

Source code in hololinked/hololinked/serializers/serializers.py

@classmethod
def for_object(cls, thing_id: str, thing_cls: str, objekt: str) -> BaseSerializer | None:
    """
    Retrieve a serializer for a given property, action or event

    Parameters
    ----------
    thing_id: str | Any
        the id of the Thing or the Thing that owns the property, action or event
    thing_cls: str | Any
        the class name of the Thing or the Thing that owns the property, action or event
    objekt: str
        the name of the property, action or event

    Returns
    -------
    BaseSerializer | JSONSerializer
        the serializer for the property, action or event. If no serializer is found, the default JSONSerializer is
        returned.
    """
    if len(cls.object_serializer_map) == 0 and len(cls.object_content_type_map) == 0:
        return cls.default
    for thing in [thing_id, thing_cls]:  # first thing id, then thing cls
        if thing in cls.object_serializer_map:
            if objekt in cls.object_serializer_map[thing]:
                return cls.object_serializer_map[thing][objekt]
        if thing in cls.object_content_type_map:
            if objekt in cls.object_content_type_map[thing]:
                return cls.content_types.get(cls.object_content_type_map[thing][objekt], None)
                # if said content type has no serializer, return None instead of default serializer
    return cls.default  # JSON is default serializer

register classmethod

register(serializer: BaseSerializer, name: str | None = None, override: bool = False) -> None

Register a new serializer. It is recommended to implement a content type property/attribute for the serializer to facilitate automatic deserialization on client side, otherwise deserialization is not gauranteed. Moreover, the said serializer must be defined on both client and server side if running in a distributed environment.

Parameters:

Name	Type	Description	Default
serializer	BaseSerializer	the serializer to register	required
name	str | None	the name of the serializer to be accessible under the object namespace. If not provided, the name of the serializer class is used.	None
override	bool	whether to override the serializer if the content type is already registered, by default False & raises ValueError for duplicate content type. For example, registering a custom JSON serializer will conflict with the default JSONSerializer, so set override=True.	False

Raises:

Type	Description
ValueError	if the serializer content type is already registered

Source code in hololinked/hololinked/serializers/serializers.py

@classmethod
def register(cls, serializer: BaseSerializer, name: str | None = None, override: bool = False) -> None:
    """
    Register a new serializer. It is recommended to implement a content type property/attribute for the serializer
    to facilitate automatic deserialization on client side, otherwise deserialization is not gauranteed.
    Moreover, the said serializer must be defined on both client and server side if running in a distributed
    environment.

    Parameters
    ----------
    serializer: BaseSerializer
        the serializer to register
    name: str, optional
        the name of the serializer to be accessible under the object namespace. If not provided, the name of the
        serializer class is used.
    override: bool, optional
        whether to override the serializer if the content type is already registered,
        by default False & raises ValueError for duplicate content type. For example, registering
        a custom JSON serializer will conflict with the default JSONSerializer, so set `override=True`.

    Raises
    ------
    ValueError
        if the serializer content type is already registered
    """
    try:
        if serializer.content_type in cls.content_types and not override:
            raise ValueError("content type already registered : {}".format(serializer.content_type))
        cls.content_types[serializer.content_type] = serializer
    except NotImplementedError:
        warnings.warn("serializer does not implement a content type", category=UserWarning)
    cls[name or serializer.__name__] = serializer

register_content_type_for_object classmethod

register_content_type_for_object(objekt: Any, content_type: str) -> None

Register content type for a property, action, event, or a Thing class to use a specific serializer. If no serializer is found, content type could still be used as metadata.

Parameters:

Name	Type	Description	Default
objekt	Any	the property, action or event. string is not accepted - use register_content_type_for_object_by_name() instead.	required
content_type	str	the content type for the value of the objekt or the serializer to be used	required

Raises:

Type	Description
ValueError	if the object is not a Property, Action or Event

Source code in hololinked/hololinked/serializers/serializers.py

@classmethod
def register_content_type_for_object(cls, objekt: Any, content_type: str) -> None:
    """
    Register content type for a property, action, event, or a `Thing` class to use a specific serializer.
    If no serializer is found, content type could still be used as metadata.

    Parameters
    ----------
    objekt: Property | Action | Event | Thing
        the property, action or event. string is not accepted - use `register_content_type_for_object_by_name()` instead.
    content_type: str
        the content type for the value of the objekt or the serializer to be used

    Raises
    ------
    ValueError
        if the object is not a Property, Action or Event
    """
    from ..core import Action, Event, Property, Thing

    if not isinstance(objekt, (Property, Action, Event)) and not issubklass(objekt, Thing):
        raise ValueError("object must be a Property, Action or Event, got : {}".format(type(objekt)))
    if issubklass(objekt, Thing):
        owner = objekt.__name__
    elif not objekt.owner:
        raise ValueError("object owner cannot be determined, cannot register content type: {}".format(objekt))
    else:
        owner = objekt.owner.__name__
    if owner not in cls.object_content_type_map:
        cls.object_content_type_map[owner] = dict()
    if issubklass(objekt, Thing):
        cls.object_content_type_map[owner][objekt.__name__] = content_type
        # its a redundant key, TODO - may be there is a better way to structure this map
    else:
        cls.object_content_type_map[owner][objekt.name] = content_type

register_for_object classmethod

register_for_object(objekt: Any, serializer: BaseSerializer) -> None

Register (an existing) serializer for a property, action or event. Other option is to register a content type, the effects are similar.

Parameters:

Name	Type	Description	Default
objekt	Any	the property, action or event	required
serializer	BaseSerializer	the serializer to be used	required

Source code in hololinked/hololinked/serializers/serializers.py

@classmethod
def register_for_object(cls, objekt: Any, serializer: BaseSerializer) -> None:
    """
    Register (an existing) serializer for a property, action or event. Other option is to register a content type,
    the effects are similar.

    Parameters
    ----------
    objekt: str | Property | Action | Event
        the property, action or event
    serializer: BaseSerializer
        the serializer to be used
    """
    if not isinstance(serializer, BaseSerializer):
        raise ValueError("serializer must be an instance of BaseSerializer, given : {}".format(type(serializer)))
    from ..core import Action, Event, Property, Thing

    if not isinstance(objekt, (Property, Action, Event)) and not issubklass(objekt, Thing):
        raise ValueError("object must be a Property, Action or Event, or Thing, got : {}".format(type(objekt)))
    if issubklass(objekt, Thing):
        owner = objekt.__name__
    elif not objekt.owner:
        raise ValueError("object owner cannot be determined : {}".format(objekt))
    else:
        owner = objekt.owner.__name__
    if owner not in cls.object_serializer_map:
        cls.object_serializer_map[owner] = dict()
    if issubklass(objekt, Thing):
        cls.object_serializer_map[owner][objekt.__name__] = serializer
    else:
        cls.object_serializer_map[owner][objekt.name] = serializer

Attributes

json: JSONSerializer

class-attribute, writable
The default serializer for all properties, actions and events (msgspec based C++ implementation)

pickle: PickleSerializer

class-attribute, writable
pickle serializer, unsafe without encryption but useful for faster & flexible serialization of python specific types

msgpack: MsgPackSerializer

class-attribute, writable
MessagePack serializer, efficient binary format that is both fast & interoperable between languages but not human readable

default: BaseSerializer

class-attribute, writable
JSONSerializer, set it to use something else