communication.serialization module

This module contains the serialization logic used in sending and receiving arbitrary objects.

exception communication.serialization.AnnotationError

Bases: Exception

Raised when an improperly function is incorrectly annotated

exception communication.serialization.RepetitionError

Bases: Exception

Raised when the action has already been performed and should not be repeated

class communication.serialization.Serialization

Bases: object

Virtual class that provides packing and unpacking functions used for communications. The outline is as follows: - serialization functions for different classes - packing function that handles metadata and determines which serialization needs to happen

• deserialization functions for different classes

• unpacking function that handles metadata and determines which deserialization needs to happen

static bitarray_deserialize(obj, **_kwargs)

Function for deserializing bitarrays

Parameters:

• obj (bytes) – object to deserialize

• **_kwargs (Any) – optional extra keyword arguments

Return type:

bitarray

Returns:

deserialized bitarray object

static bitarray_serialize(obj, **_kwargs)

Function for serializing bitarray

Parameters:

• obj (bitarray) – bitarray object to serialize

• **_kwargs (Any) – optional extra keyword arguments

Return type:

bytes

Returns:

serialized object

static clear_new_serialization_logic()

Clear all custom serialization (and deserialization) logic that was added to this class.

Return type:

None

static collection_deserialize(collection_obj, **kwargs)

Function for deserializing collections

Parameters:

• collection_obj (Union[List[Any], Dict[str, Any]]) – object to deserialize

• **kwargs (Any) – optional extra keyword arguments

Raises:

ValueError – raised when (nested) value cannot be deserialized

Return type:

Union[Dict[str, Any], List[Any]]

Returns:

deserialized collection

custom_deserialization_funcs: ClassVar[Dict[str, Callable[[Any, Any], SupportsSerialization]]] = {}

custom_serialization_funcs: ClassVar[Dict[str, Callable[[SupportsSerialization, Any], Any]]] = {}

static default_deserialize(obj, use_pickle=False, **_kwargs)

Fall-back function is case no specific deserialization function is available. This function uses the pickle library

Parameters:

• obj (bytes) – object to deserialize

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback deserializer

• **_kwargs (Any) – optional extra keyword arguments

Return type:

Any

Returns:

deserialized object

static default_serialize(obj, use_pickle, **_kwargs)

Fall-back function is case no specific serialization function is available. This function uses the pickle library

Parameters:

• obj (Any) – object to serialize

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback serializer

• **_kwargs (Any) – optional extra keyword arguments

Raises:

NotImplementedError – raised when no serialization function is defined for object

Return type:

bytes

Returns:

serialized object

static deserialize(obj, use_pickle=False, **kwargs)

Function that detects which deserialization function should be run and calls it

Parameters:

• obj (Any) – object to deserialize

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback deserializer

• **kwargs (Any) – optional extra keyword arguments

Return type:

Any

Returns:

deserialized object

static gmpy_deserialize(obj, **_kwargs)

Function for deserializing gmpy objects

Parameters:

• obj – object to deserialize

• **_kwargs – optional extra keyword arguments

Returns:

deserialized gmpy object

static gmpy_serialize(obj, **_kwargs)

Function for serializing gmpy objects

Parameters:

• obj – gmpy object to serialize

• **_kwargs – optional extra keyword arguments

Returns:

serialized object

static int_deserialize(obj, **_kwargs)

Function for deserializing Python ints

Parameters:

• obj (bytes) – object to deserialize

• **_kwargs (Any) – optional extra keyword arguments

Return type:

int

Returns:

deserialized int object

static int_serialize(obj, **_kwargs)

Function for serializing Python ints

Parameters:

• obj (int) – int object to serialize

• **_kwargs (Any) – optional extra keyword arguments

Return type:

bytes

Returns:

serialized object

static numpy_deserialize(obj, **kwargs)

Function for serializing numpy object arrays

Parameters:

• obj (Dict[str, List[Any]]) – numpy object to serialize

• **kwargs (Any) – optional extra keyword arguments

Return type:

ndarray[tuple[Any, ...], dtype[object_]]

Returns:

deserialized object

static numpy_serialize(obj, **_kwargs)

Function for serializing numpy object arrays

Parameters:

• obj (ndarray[tuple[Any, ...], dtype[Any]]) – numpy object to serialize

• **_kwargs (Any) – optional extra keyword arguments

Return type:

Dict[str, List[Any]]

Returns:

serialized object

static pack(obj, msg_id, use_pickle, option=392, **kwargs)

Function that adds metadata and serializes the object for transmission.

Parameters:

• obj (Any) – object to pack

• msg_id (Union[str, int]) – message identifier associated to the message

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback packer

• option (Optional[int]) – ormsgpack options can be specified through this parameter

• **kwargs (Any) – optional extra keyword arguments

Return type:

bytes

Returns:

packed object (serialized and annotated)

static serialize(obj, use_pickle, **kwargs)

Function that detects with serialization function should be used and applies it

Parameters:

• obj (Any) – object to serialize

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback serializer

• **kwargs (Any) – optional extra keyword arguments

Return type:

Union[bytes, Dict[str, bytes]]

Returns:

serialized object

static set_serialization_logic(obj_class, check_annotations=True)

Function for storing serialization logic for classes that have not been specified here or need to be overridden

Parameters:

• obj_class (Type[SupportsSerialization]) – object class to set serialization logic for

• check_annotations (bool) – validate return annotation of the serialization logic

Raises:

• RepetitionError – raised when serialization function is already defined for object class

• TypeError – raised when provided object class has no (de)serialization function

• AnnotationError – raised when the return annotation is inconsistent

Return type:

None

static tuple_deserialize(obj, **kwargs)

Function for deserializing tuples

Parameters:

• obj (List[Any]) – object to deserialize

• **kwargs (Any) – optional extra keyword arguments

Return type:

Tuple[Any, ...]

Returns:

deserialized tuple object

static tuple_serialize(obj, **_kwargs)

Function for serializing tuples

Parameters:

• obj (Tuple[Any, ...]) – tuple object to serialize

• **_kwargs (Any) – optional extra keyword arguments

Return type:

List[Any]

Returns:

serialized object

static unpack(obj, use_pickle=False, option=None, **kwargs)

Function that handles metadata and turns the bytes object into a python object

Parameters:

• obj (bytes) – bytes object to unpack

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback deserializer

• option (Optional[int]) – ormsgpack options can be specified through this parameter

• **kwargs (Any) – optional extra keyword arguments

Return type:

Tuple[str, Any]

Returns:

unpacked object

class communication.serialization.SupportsSerialization(*args, **kwargs)

Bases: Protocol

Type placeholder for classes supporting custom serialization.

static deserialize(obj, **kwargs)

Deserialize the given object into an object of this class.

Parameters:

• obj (Any) – object to be deserialized.

• **kwargs (Any) – Optional extra keyword arguments.

Return type:

SupportsSerialization

Returns:

Deserialized object.

serialize(**kwargs)

Serialize this object into bytes.

Parameters:

**kwargs (Any) – Optional extra keyword arguments.

Return type:

Any

Returns:

Serialization of this instance to Dict with bytes.