academy.serialize

academy/serialize.py

default_serializer module-attribute

default_serializer: ContextVar[SerializationStrategy] = (
    ContextVar("default_serializer", default=PICKLE)
)

Default serialization method used to send requests.

allowed_deserializers module-attribute

allowed_deserializers: ContextVar[
    set[SerializationStrategy]
] = ContextVar(
    "allowed_deserializers", default=ALL_SERIALIZERS
)

Deserializers allowed in this context.

NoPickleMixin

Mixin that raises an error if a type is pickled.

Serializer

Bases: Protocol

Protocol for different types of serialization methods.

serialize classmethod

serialize(obj: Any) -> str

Serialize object.

Source code in academy/serialize.py

@classmethod
def serialize(cls, obj: Any) -> str:
    """Serialize object."""
    ...

deserialize classmethod

deserialize(data: str) -> Any

Deserialize data.

Source code in academy/serialize.py

@classmethod
def deserialize(cls, data: str) -> Any:
    """Deserialize data."""
    ...

JsonSerializer

Serializes objects into json.

Works only on basic python types, but allows deserialization without arbitrary code execution, and is compatible between python versions.

serialize classmethod

serialize(obj: Any) -> str

Serialize object.

Source code in academy/serialize.py

@classmethod
def serialize(cls, obj: Any) -> str:
    """Serialize object."""
    return json.dumps(obj, default=json_exception_serializer)

deserialize classmethod

deserialize(data: str) -> Any

Deserialize data.

Source code in academy/serialize.py

@classmethod
def deserialize(cls, data: str) -> Any:
    """Deserialize data."""
    return json.loads(data, object_hook=json_exception_deserializer)

PickleSerializer

Serializes objects using pickle.

Seriliazes objects using pickle. Generally allows for a wider range of types to be serialized, but cannot deserialize without allowing arbitrary code execution. Sending pickled python objects between processes with different python versions or different environments can lead to unexpected results.

serialize classmethod

serialize(obj: Any) -> str

Serialize object.

Source code in academy/serialize.py

@classmethod
def serialize(cls, obj: Any) -> str:
    """Serialize object."""
    raw = pickle.dumps(obj)
    return base64.b64encode(raw).decode('utf-8')

deserialize classmethod

deserialize(data: str) -> Any

Deserialize data.

Source code in academy/serialize.py

@classmethod
def deserialize(cls, data: str) -> Any:
    """Deserialize data."""
    return pickle.loads(base64.b64decode(data))

SerializationStrategy

Bases: str, Enum

Enum for different serialization strategies.

json_exception_serializer

json_exception_serializer(
    obj: Any,
) -> dict[str, bool | str]

Serialize exceptions as json.

Extension to json serialization to deal with arbitrary exceptions.

Source code in academy/serialize.py

def json_exception_serializer(obj: Any) -> dict[str, bool | str]:
    """Serialize exceptions as json.

    Extension to json serialization to deal with arbitrary exceptions.
    """
    if isinstance(obj, BaseException):
        exception_str = ''.join(
            traceback.format_exception(
                type(obj),
                obj,
                obj.__traceback__,
            ),
        )
        return {
            '__exception__': True,
            'exception_str': exception_str,
        }

    raise TypeError(
        f'Cannot serialize object of {type(obj)}',
    )  # pragma: no cover

json_exception_deserializer

json_exception_deserializer(
    dct: dict[str, Any],
) -> Exception | dict[str, Any]

Deserialize exceptions from json.

Extension to json serializer that parses serialized exceptions into a RemoteException.

Source code in academy/serialize.py

def json_exception_deserializer(
    dct: dict[str, Any],
) -> Exception | dict[str, Any]:
    """Deserialize exceptions from json.

    Extension to json serializer that parses serialized exceptions into
    a RemoteException.
    """
    if '__exception__' in dct:
        return AcademyRemoteError(dct['exception_str'])

    return dct

serialize

serialize(obj: Any, strategy: SerializationStrategy) -> str

Serialize object using strategy.

Parameters:

• obj (Any) –

  Object to be serialized.

• strategy (SerializationStrategy) –

  Strategy used for serialization.

Returns:

• str –

  Serialized data.

Source code in academy/serialize.py

def serialize(obj: Any, strategy: SerializationStrategy) -> str:
    """Serialize object using strategy.

    Args:
        obj: Object to be serialized.
        strategy: Strategy used for serialization.

    Returns:
        Serialized data.
    """
    serializer = _get_serializer(strategy)
    return serializer.serialize(obj)

deserialize

deserialize(
    obj: str, strategy: SerializationStrategy
) -> Any

Deserialize object using strategy.

Parameters:

• obj (str) –

  Json data.

• strategy (SerializationStrategy) –

  Strategy used for serialization.

Returns:

• Any –

  Deserialized object.

Source code in academy/serialize.py

def deserialize(obj: str, strategy: SerializationStrategy) -> Any:
    """Deserialize object using strategy.

    Args:
        obj: Json data.
        strategy: Strategy used for serialization.

    Returns:
        Deserialized object.
    """
    if strategy not in allowed_deserializers.get():
        raise DeserializationMethodProhibitedError()
    serializer = _get_serializer(strategy)
    return serializer.deserialize(obj)