hololinked.serializers.serializers.Serializers

A singleton class that holds all serializers and provides a registry for content types. All members are class attributes.

• For property, the value is serialized using the serializer registered for the property.
• For action, the return value is serialized using the serializer registered for the action.
• For event, the payload is serialized using the serializer registered for the event.

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\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. 

    - For property, the value is serialized using the serializer registered for the property.
    - For action, the return value is serialized using the serializer registered for the action.
    - For event, the payload is serialized using the serializer registered for the event.

    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
    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
    msgpack = ClassSelector(default=MsgpackSerializer(), class_=BaseSerializer, class_member=True, 
                        doc="MessagePack serializer, efficient binary format that is both fast & interoperable between languages ") # type: BaseSerializer
    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
    default = ClassSelector(default=json.default, class_=BaseSerializer, class_member=True, 
                        doc="The default serialization to be used") # type: BaseSerializer
    default_content_type = String(default=default.default.content_type, class_member=True,
                        doc="The default content type for the default serializer") # type: str

    content_types = Parameter(default={
                                'application/json': json.default,
                                'application/octet-stream': pickle.default,
                                'x-msgpack': msgpack.default,
                                'text/plain': text.default
                            }, doc="A dictionary of content types and their serializers",
                            readonly=True, class_member=True) # type: typing.Dict[str, BaseSerializer]
    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: typing.Dict[str, typing.Dict[str, str]]
    object_serializer_map = Parameter(default=dict(), class_member=True, 
                                doc="A dictionary of serializer for specific properties, actions and events",
                                readonly=True) # type: typing.Dict[str, typing.Dict[str, BaseSerializer]]
    protocol_serializer_map = Parameter(default=dict(), class_member=True, 
                                doc="A dictionary of serializer for a specific protocol",
                                readonly=True) # type: typing.Dict[str, BaseSerializer]


    @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 by this package.

        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. 

        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:
        """
        Retrieve a serializer for a given property, action or event

        Parameters
        ----------
        thing: str | typing.Any
            the id of the Thing or the Thing that owns the property, action or event
        objekt: str | Property | Action | Event
            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[cls.object_content_type_map[thing][objekt]]
        return cls.default # JSON is default serializer


    # @validate_call
    @classmethod
    def register_content_type_for_object(cls, objekt: typing.Any, content_type: str) -> None:
        """
        Register content type for a property, action, event, or a `Thing` class to use a specific serializer.

        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
        """
        if content_type not in cls.content_types:
            raise ValueError("content type {} unsupported".format(content_type))
        from ..core import Property, Action, Event, 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] = content_type
        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 | typing.Any, content_type: str) -> None:
        """
        Register an existing 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.

        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
        """
        if not content_type in cls.content_types:
            raise ValueError("content type {} unsupported".format(content_type))
        from ..core import Property, Action, Event
        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
        """
        if content_type not in cls.content_types:
            raise ValueError("content type {} unsupported".format(content_type))
        cls.object_content_type_map[thing_id] = content_type

    # @validate_call
    @classmethod
    def register_for_object(cls, objekt: typing.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 Property, Action, Event, 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] = serializer
        else:
            cls.object_serializer_map[owner][objekt.name] = serializer

    @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. 

        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
        """
        cls.object_serializer_map[thing_id] = serializer

Functions

for_object classmethod

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

Retrieve a serializer for a given property, action or event

Parameters:

Name	Type	Description	Default
thing		the id 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\serializers\serializers.py

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

    Parameters
    ----------
    thing: str | typing.Any
        the id of the Thing or the Thing that owns the property, action or event
    objekt: str | Property | Action | Event
        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[cls.object_content_type_map[thing][objekt]]
    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 by this package.

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.	False

Raises:

Type	Description
ValueError	if the serializer content type is already registered

Source code in 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 by this package.

    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. 

    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: typing.Any, content_type: str) -> None

Register content type for a property, action, event, or a Thing class to use a specific serializer.

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\serializers\serializers.py

@classmethod
def register_content_type_for_object(cls, objekt: typing.Any, content_type: str) -> None:
    """
    Register content type for a property, action, event, or a `Thing` class to use a specific serializer.

    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
    """
    if content_type not in cls.content_types:
        raise ValueError("content type {} unsupported".format(content_type))
    from ..core import Property, Action, Event, 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] = content_type
    else:
        cls.object_content_type_map[owner][objekt.name] = content_type    

register_for_object classmethod

register_for_object(objekt: typing.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\serializers\serializers.py

@classmethod
def register_for_object(cls, objekt: typing.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 Property, Action, Event, 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] = 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