Skip to content

laila.entry

laila.entry

Entry sub-package providing the core Entry class and pre-built transformation sequences.

TransformationSequence

Bases: BaseModel

Pydantic v2 sequential transformation pipeline.

  • forward(): applies transformations in order
  • backward(): applies transformations in reverse order
  • empty pipeline = identity (no-op)
Source code in entry/compdata/transformation/base.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class TransformationSequence(BaseModel):
    """
    Pydantic v2 sequential transformation pipeline.

    - forward(): applies transformations in order
    - backward(): applies transformations in reverse order
    - empty pipeline = identity (no-op)
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    transformations: List[_data_transformation] = Field(default_factory=list)

    def forward(self, data: Any) -> Any:
        """Apply all transformations in order and collect inverse codes.

        Parameters
        ----------
        data : Any
            Input data to transform.

        Returns
        -------
        tuple[Any, list[str]]
            Transformed data and the list of inverse code snippets in
            reverse order.
        """
        current = data
        inverse_codes: List[str] = []

        for transformation in self.transformations:
            current = transformation.forward(current)
            inverse_codes.append(transformation.backward_code)

        return current, inverse_codes[::-1]


    def __iter__(self):
        """Iterate over the contained transformations."""
        return iter(self.transformations)


    def append(self, t) -> Any:
        """Append one or more transformations to the pipeline.

        Parameters
        ----------
        t : _data_transformation or list[_data_transformation]
            Transformation(s) to append.

        Returns
        -------
        TransformationSequence
            ``self``, for chaining.

        Raises
        ------
        TypeError
            If *t* is not a valid transformation or list thereof.
        """
        if isinstance (t, _data_transformation):
            self.transformations.append(t)
            return self

        if isinstance (t, list):
            for _t in t:
                if not isinstance(_t, _data_transformation):
                    raise TypeError("All elements must be _data_transformation instances.")
                self.transformations.append(_t)
            return self

        raise TypeError("append expects a _data_transformation or list of them.")


    def __repr__(self) -> str:
        """Return a human-readable pipeline summary."""
        if not self.transformations:
            return f"{self.__class__.__name__}(identity)"
        names = " -> ".join(getattr(t, "name", t.__class__.__name__) for t in self.transformations)
        return f"{self.__class__.__name__}({names})"

forward(data)

Apply all transformations in order and collect inverse codes.

Parameters:

Name Type Description Default
data Any

Input data to transform.

 required

Returns:

Type Description
tuple[Any, list[str]]

Transformed data and the list of inverse code snippets in reverse order.
Source code in entry/compdata/transformation/base.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def forward(self, data: Any) -> Any:
    """Apply all transformations in order and collect inverse codes.

    Parameters
    ----------
    data : Any
        Input data to transform.

    Returns
    -------
    tuple[Any, list[str]]
        Transformed data and the list of inverse code snippets in
        reverse order.
    """
    current = data
    inverse_codes: List[str] = []

    for transformation in self.transformations:
        current = transformation.forward(current)
        inverse_codes.append(transformation.backward_code)

    return current, inverse_codes[::-1]

__iter__()

Iterate over the contained transformations.

Source code in entry/compdata/transformation/base.py 
76
77
78
def __iter__(self):
    """Iterate over the contained transformations."""
    return iter(self.transformations)

append(t)

Append one or more transformations to the pipeline.

Parameters:

Name Type Description Default
t _data_transformation or list[_data_transformation]

Transformation(s) to append.

 required

Returns:

Type Description
TransformationSequence

self, for chaining.

Raises:

Type Description
TypeError

If t is not a valid transformation or list thereof.
Source code in entry/compdata/transformation/base.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def append(self, t) -> Any:
    """Append one or more transformations to the pipeline.

    Parameters
    ----------
    t : _data_transformation or list[_data_transformation]
        Transformation(s) to append.

    Returns
    -------
    TransformationSequence
        ``self``, for chaining.

    Raises
    ------
    TypeError
        If *t* is not a valid transformation or list thereof.
    """
    if isinstance (t, _data_transformation):
        self.transformations.append(t)
        return self

    if isinstance (t, list):
        for _t in t:
            if not isinstance(_t, _data_transformation):
                raise TypeError("All elements must be _data_transformation instances.")
            self.transformations.append(_t)
        return self

    raise TypeError("append expects a _data_transformation or list of them.")

__repr__()

Return a human-readable pipeline summary.

Source code in entry/compdata/transformation/base.py 
113
114
115
116
117
118
def __repr__(self) -> str:
    """Return a human-readable pipeline summary."""
    if not self.transformations:
        return f"{self.__class__.__name__}(identity)"
    names = " -> ".join(getattr(t, "name", t.__class__.__name__) for t in self.transformations)
    return f"{self.__class__.__name__}({names})"

Base64

Bases: _data_transformation

Reversible Base64 encoding transformation.

Forward encodes binary data to a Base64 UTF-8 string; backward decodes it back to raw bytes.

Source code in entry/compdata/transformation/base64/base64.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class Base64(_data_transformation):
    """Reversible Base64 encoding transformation.

    Forward encodes binary data to a Base64 UTF-8 string;
    backward decodes it back to raw bytes.
    """

    name: str = "base64"

    def model_post_init(self, __context: Any) -> None:
        """Build standalone backward recovery code."""
        # Standalone recovery code with embedded kwargs (e.g., altchars, validate)
        self.backward_code = (
            "def backward(inp):\n"
            "    import base64\n"
            f"    kwargs = {self.backward_kwargs!r}\n"
            "    if isinstance(inp, memoryview):\n"
            "        inp = inp.tobytes()\n"
            "    return base64.b64decode(inp, **kwargs)\n"
        )

    def forward(self, data: Union[bytes, bytearray, memoryview]) -> str:
        """Encode binary data -> Base64 UTF-8 string."""
        if isinstance(data, memoryview):
            data = data.tobytes()
        if not isinstance(data, (bytes, bytearray)):
            raise TypeError("Base64.forward expects bytes/bytearray/memoryview")
        # b64encode supports altchars=...
        out = base64.b64encode(bytes(data), **self.forward_kwargs)
        return out.decode("utf-8")

    def backward(self, payload: Union[str, bytes, bytearray, memoryview]) -> bytes:
        """Decode Base64 (str/bytes/bytearray/memoryview) -> raw bytes."""
        if isinstance(payload, memoryview):
            payload = payload.tobytes()
        # base64.b64decode accepts str or bytes-like
        try:
            return base64.b64decode(payload, **self.backward_kwargs)
        except Exception as e:
            raise ValueError("Invalid Base64 payload") from e

model_post_init(__context)

Build standalone backward recovery code.

Source code in entry/compdata/transformation/base64/base64.py 
17
18
19
20
21
22
23
24
25
26
27
def model_post_init(self, __context: Any) -> None:
    """Build standalone backward recovery code."""
    # Standalone recovery code with embedded kwargs (e.g., altchars, validate)
    self.backward_code = (
        "def backward(inp):\n"
        "    import base64\n"
        f"    kwargs = {self.backward_kwargs!r}\n"
        "    if isinstance(inp, memoryview):\n"
        "        inp = inp.tobytes()\n"
        "    return base64.b64decode(inp, **kwargs)\n"
    )

forward(data)

Encode binary data -> Base64 UTF-8 string.

Source code in entry/compdata/transformation/base64/base64.py 
29
30
31
32
33
34
35
36
37
def forward(self, data: Union[bytes, bytearray, memoryview]) -> str:
    """Encode binary data -> Base64 UTF-8 string."""
    if isinstance(data, memoryview):
        data = data.tobytes()
    if not isinstance(data, (bytes, bytearray)):
        raise TypeError("Base64.forward expects bytes/bytearray/memoryview")
    # b64encode supports altchars=...
    out = base64.b64encode(bytes(data), **self.forward_kwargs)
    return out.decode("utf-8")

backward(payload)

Decode Base64 (str/bytes/bytearray/memoryview) -> raw bytes.

Source code in entry/compdata/transformation/base64/base64.py 
39
40
41
42
43
44
45
46
47
def backward(self, payload: Union[str, bytes, bytearray, memoryview]) -> bytes:
    """Decode Base64 (str/bytes/bytearray/memoryview) -> raw bytes."""
    if isinstance(payload, memoryview):
        payload = payload.tobytes()
    # base64.b64decode accepts str or bytes-like
    try:
        return base64.b64decode(payload, **self.backward_kwargs)
    except Exception as e:
        raise ValueError("Invalid Base64 payload") from e

Zlib

Bases: _data_transformation

Reversible zlib compression transformation.

Forward compresses a UTF-8 string to a Base64-encoded compressed string; backward decompresses it.

Source code in entry/compdata/transformation/compression/zlib.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class Zlib(_data_transformation):
    """Reversible zlib compression transformation.

    Forward compresses a UTF-8 string to a Base64-encoded compressed string;
    backward decompresses it.
    """

    name: str = "zlib"

    def model_post_init(self, __context: Any) -> None:
        """Build standalone backward recovery code."""
        # Standalone recovery code mirroring `backward()` with embedded kwargs
        self.backward_code = textwrap.dedent(f"""
            def backward(data):
                import zlib, base64
                if not isinstance(data, str):
                    raise TypeError("Zlib.backward expects a Base64 string (str)")
                compressed = base64.b64decode(data, validate=True)
                return zlib.decompress(compressed, **{self.backward_kwargs!r}).decode("utf-8")
        """)

    def forward(self, data: str) -> str:
        """Compress a UTF-8 string and return a Base64-encoded result.

        Parameters
        ----------
        data : str
            Plain-text UTF-8 string.

        Returns
        -------
        str
            Base64-encoded compressed bytes.

        Raises
        ------
        TypeError
            If *data* is not a ``str``.
        """
        if not isinstance(data, str):
            raise TypeError("Zlib.forward expects a UTF-8 string (str)")
        compressed = zlib.compress(data.encode("utf-8"), **self.forward_kwargs)
        return base64.b64encode(compressed).decode("utf-8")

    def backward(self, data: str) -> str:
        """Decompress a Base64-encoded compressed string.

        Parameters
        ----------
        data : str
            Base64-encoded compressed payload.

        Returns
        -------
        str
            Original UTF-8 string.

        Raises
        ------
        TypeError
            If *data* is not a ``str``.
        """
        if not isinstance(data, str):
            raise TypeError("Zlib.backward expects a Base64 string (str)")
        compressed = base64.b64decode(data, validate=True)
        return zlib.decompress(compressed, **self.backward_kwargs).decode("utf-8")

model_post_init(__context)

Build standalone backward recovery code.

Source code in entry/compdata/transformation/compression/zlib.py 
19
20
21
22
23
24
25
26
27
28
29
def model_post_init(self, __context: Any) -> None:
    """Build standalone backward recovery code."""
    # Standalone recovery code mirroring `backward()` with embedded kwargs
    self.backward_code = textwrap.dedent(f"""
        def backward(data):
            import zlib, base64
            if not isinstance(data, str):
                raise TypeError("Zlib.backward expects a Base64 string (str)")
            compressed = base64.b64decode(data, validate=True)
            return zlib.decompress(compressed, **{self.backward_kwargs!r}).decode("utf-8")
    """)

forward(data)

Compress a UTF-8 string and return a Base64-encoded result.

Parameters:

Name Type Description Default
data str

Plain-text UTF-8 string.

 required

Returns:

Type Description
str

Base64-encoded compressed bytes.

Raises:

Type Description
TypeError

If data is not a str.
Source code in entry/compdata/transformation/compression/zlib.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def forward(self, data: str) -> str:
    """Compress a UTF-8 string and return a Base64-encoded result.

    Parameters
    ----------
    data : str
        Plain-text UTF-8 string.

    Returns
    -------
    str
        Base64-encoded compressed bytes.

    Raises
    ------
    TypeError
        If *data* is not a ``str``.
    """
    if not isinstance(data, str):
        raise TypeError("Zlib.forward expects a UTF-8 string (str)")
    compressed = zlib.compress(data.encode("utf-8"), **self.forward_kwargs)
    return base64.b64encode(compressed).decode("utf-8")

backward(data)

Decompress a Base64-encoded compressed string.

Parameters:

Name Type Description Default
data str

Base64-encoded compressed payload.

 required

Returns:

Type Description
str

Original UTF-8 string.

Raises:

Type Description
TypeError

If data is not a str.
Source code in entry/compdata/transformation/compression/zlib.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def backward(self, data: str) -> str:
    """Decompress a Base64-encoded compressed string.

    Parameters
    ----------
    data : str
        Base64-encoded compressed payload.

    Returns
    -------
    str
        Original UTF-8 string.

    Raises
    ------
    TypeError
        If *data* is not a ``str``.
    """
    if not isinstance(data, str):
        raise TypeError("Zlib.backward expects a Base64 string (str)")
    compressed = base64.b64decode(data, validate=True)
    return zlib.decompress(compressed, **self.backward_kwargs).decode("utf-8")

FernetEncryption

Bases: _data_transformation

Reversible Fernet symmetric encryption transformation.

Parameters:

Name Type Description Default
key str or bytes

Fernet-compatible encryption key.

 required
Source code in entry/compdata/transformation/encryption/encryption.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class FernetEncryption(_data_transformation):
    """Reversible Fernet symmetric encryption transformation.

    Parameters
    ----------
    key : str or bytes
        Fernet-compatible encryption key.
    """

    name: str = "fernet"
    key: Union[str, bytes]
    _fernet: Any = None

    @field_validator("key")
    @classmethod
    def _coerce_key(cls, v: Union[str, bytes]) -> bytes:
        """Coerce string keys to bytes."""
        if isinstance(v, str):
            v = v.encode("utf-8")
        if not isinstance(v, (bytes, bytearray)):
            raise TypeError("Encryption.key must be bytes or str")
        return bytes(v)

    def model_post_init(self, __context: Any) -> None:
        """Initialise the Fernet encryptor and build backward recovery code."""
        from cryptography.fernet import Fernet
        self._fernet = Fernet(self.key)

        # build standalone recovery code (uses optional ttl from backward_kwargs)
        self.backward_code = textwrap.dedent(f"""
            def backward(inp):
                from cryptography.fernet import Fernet, InvalidToken
                if not isinstance(inp, str):
                    raise TypeError("Encryption.backward expects a Fernet token string (str)")
                f = Fernet(key={self.key!r})
                token = inp.encode("utf-8")
                kwargs = {self.backward_kwargs!r}
                ttl = kwargs.get("ttl", None)
                try:
                    out = f.decrypt(token, ttl=ttl)
                except InvalidToken as e:
                    raise ValueError("Invalid Fernet token or TTL expired") from e
                return out.decode("utf-8")
        """)

    def forward(self, data: str) -> str:
        """Encrypt a UTF-8 string and return the Fernet token as a string.

        Parameters
        ----------
        data : str
            Plain-text string to encrypt.

        Returns
        -------
        str
            Fernet token.

        Raises
        ------
        TypeError
            If *data* is not a ``str``.
        """
        if not isinstance(data, str):
            raise TypeError("Encryption.forward expects a Base64 string (str)")
        return self._fernet.encrypt(data.encode("utf-8")).decode("utf-8")

    def backward(self, data: str) -> str:
        """Decrypt a Fernet token back to the original string.

        Parameters
        ----------
        data : str
            Fernet token string.

        Returns
        -------
        str
            Original plain-text string.

        Raises
        ------
        TypeError
            If *data* is not a ``str``.
        ValueError
            If the token is invalid or TTL has expired.
        """
        if not isinstance(data, str):
            raise TypeError("Encryption.backward expects a Fernet token string (str)")
        ttl = self.backward_kwargs.get("ttl", None)
        from cryptography.fernet import InvalidToken
        try:
            return self._fernet.decrypt(data.encode("utf-8"), ttl=ttl).decode("utf-8")
        except InvalidToken as e:
            raise ValueError("Invalid Fernet token or TTL expired") from e

model_post_init(__context)

Initialise the Fernet encryptor and build backward recovery code.

Source code in entry/compdata/transformation/encryption/encryption.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def model_post_init(self, __context: Any) -> None:
    """Initialise the Fernet encryptor and build backward recovery code."""
    from cryptography.fernet import Fernet
    self._fernet = Fernet(self.key)

    # build standalone recovery code (uses optional ttl from backward_kwargs)
    self.backward_code = textwrap.dedent(f"""
        def backward(inp):
            from cryptography.fernet import Fernet, InvalidToken
            if not isinstance(inp, str):
                raise TypeError("Encryption.backward expects a Fernet token string (str)")
            f = Fernet(key={self.key!r})
            token = inp.encode("utf-8")
            kwargs = {self.backward_kwargs!r}
            ttl = kwargs.get("ttl", None)
            try:
                out = f.decrypt(token, ttl=ttl)
            except InvalidToken as e:
                raise ValueError("Invalid Fernet token or TTL expired") from e
            return out.decode("utf-8")
    """)

forward(data)

Encrypt a UTF-8 string and return the Fernet token as a string.

Parameters:

Name Type Description Default
data str

Plain-text string to encrypt.

 required

Returns:

Type Description
str

Fernet token.

Raises:

Type Description
TypeError

If data is not a str.
Source code in entry/compdata/transformation/encryption/encryption.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def forward(self, data: str) -> str:
    """Encrypt a UTF-8 string and return the Fernet token as a string.

    Parameters
    ----------
    data : str
        Plain-text string to encrypt.

    Returns
    -------
    str
        Fernet token.

    Raises
    ------
    TypeError
        If *data* is not a ``str``.
    """
    if not isinstance(data, str):
        raise TypeError("Encryption.forward expects a Base64 string (str)")
    return self._fernet.encrypt(data.encode("utf-8")).decode("utf-8")

backward(data)

Decrypt a Fernet token back to the original string.

Parameters:

Name Type Description Default
data str

Fernet token string.

 required

Returns:

Type Description
str

Original plain-text string.

Raises:

Type Description
TypeError

If data is not a str.
ValueError

If the token is invalid or TTL has expired.
Source code in entry/compdata/transformation/encryption/encryption.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def backward(self, data: str) -> str:
    """Decrypt a Fernet token back to the original string.

    Parameters
    ----------
    data : str
        Fernet token string.

    Returns
    -------
    str
        Original plain-text string.

    Raises
    ------
    TypeError
        If *data* is not a ``str``.
    ValueError
        If the token is invalid or TTL has expired.
    """
    if not isinstance(data, str):
        raise TypeError("Encryption.backward expects a Fernet token string (str)")
    ttl = self.backward_kwargs.get("ttl", None)
    from cryptography.fernet import InvalidToken
    try:
        return self._fernet.decrypt(data.encode("utf-8"), ttl=ttl).decode("utf-8")
    except InvalidToken as e:
        raise ValueError("Invalid Fernet token or TTL expired") from e

PickleSerializer

Bases: _data_transformation

Reversible pickle serialiser for arbitrary Python objects.

Source code in entry/compdata/transformation/serialization/pickle.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class PickleSerializer(_data_transformation):
    """Reversible pickle serialiser for arbitrary Python objects."""

    name: str = "pickle"

    def model_post_init(self, __context: Any) -> None:
        """Build backward_code dynamically after model creation."""
        self.backward_code = textwrap.dedent(f"""
            def backward(inp):
                import pickle
                kwargs = {self.backward_kwargs!r}
                return pickle.loads(inp, **kwargs)
        """)

    def forward(self, inp: Any) -> bytes:
        """Serialize a Python object into bytes using pickle.

        Parameters
        ----------
        inp : Any
            Object to serialize.

        Returns
        -------
        bytes
            Pickled bytes.
        """
        kwargs = {**self.forward_kwargs}
        return pickle.dumps(inp, **kwargs)

    def backward(self, inp: bytes) -> Any:
        """Deserialize bytes back into a Python object using pickle.

        Parameters
        ----------
        inp : bytes
            Pickled bytes.

        Returns
        -------
        Any
            Unpickled Python object.
        """
        kwargs = {**self.backward_kwargs}
        return pickle.loads(inp, **kwargs)

model_post_init(__context)

Build backward_code dynamically after model creation.

Source code in entry/compdata/transformation/serialization/pickle.py 
14
15
16
17
18
19
20
21
def model_post_init(self, __context: Any) -> None:
    """Build backward_code dynamically after model creation."""
    self.backward_code = textwrap.dedent(f"""
        def backward(inp):
            import pickle
            kwargs = {self.backward_kwargs!r}
            return pickle.loads(inp, **kwargs)
    """)

forward(inp)

Serialize a Python object into bytes using pickle.

Parameters:

Name Type Description Default
inp Any

Object to serialize.

 required

Returns:

Type Description
bytes

Pickled bytes.
Source code in entry/compdata/transformation/serialization/pickle.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def forward(self, inp: Any) -> bytes:
    """Serialize a Python object into bytes using pickle.

    Parameters
    ----------
    inp : Any
        Object to serialize.

    Returns
    -------
    bytes
        Pickled bytes.
    """
    kwargs = {**self.forward_kwargs}
    return pickle.dumps(inp, **kwargs)

backward(inp)

Deserialize bytes back into a Python object using pickle.

Parameters:

Name Type Description Default
inp bytes

Pickled bytes.

 required

Returns:

Type Description
Any

Unpickled Python object.
Source code in entry/compdata/transformation/serialization/pickle.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def backward(self, inp: bytes) -> Any:
    """Deserialize bytes back into a Python object using pickle.

    Parameters
    ----------
    inp : bytes
        Pickled bytes.

    Returns
    -------
    Any
        Unpickled Python object.
    """
    kwargs = {**self.backward_kwargs}
    return pickle.loads(inp, **kwargs)

MsgpackSerializer

Bases: _data_transformation

Reversible msgpack serialiser for Python objects.

Source code in entry/compdata/transformation/serialization/msgpack.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class MsgpackSerializer(_data_transformation):
    """Reversible msgpack serialiser for Python objects."""

    name: str = "msgpack"

    def model_post_init(self, __context: Any) -> None:
        """Build backward_code dynamically after model creation."""
        self.backward_code = textwrap.dedent(f"""
            def backward(inp):
                import msgpack
                kwargs = {{"raw": False, **{self.backward_kwargs!r}}}
                return msgpack.unpackb(inp, **kwargs)
        """)

    def forward(self, inp: Any) -> bytes:
        """Serialize a Python object into bytes using msgpack.

        Parameters
        ----------
        inp : Any
            Object to serialize.

        Returns
        -------
        bytes
            Msgpack-encoded bytes.
        """
        kwargs = {"use_bin_type": True, **self.forward_kwargs}
        return msgpack.packb(inp, **kwargs)

    def backward(self, inp: bytes) -> Any:
        """Deserialize bytes back into a Python object using msgpack.

        Parameters
        ----------
        inp : bytes
            Msgpack-encoded bytes.

        Returns
        -------
        Any
            Deserialized Python object.
        """
        kwargs = {"raw": False, **self.backward_kwargs}
        return msgpack.unpackb(inp, **kwargs)

model_post_init(__context)

Build backward_code dynamically after model creation.

Source code in entry/compdata/transformation/serialization/msgpack.py 
14
15
16
17
18
19
20
21
def model_post_init(self, __context: Any) -> None:
    """Build backward_code dynamically after model creation."""
    self.backward_code = textwrap.dedent(f"""
        def backward(inp):
            import msgpack
            kwargs = {{"raw": False, **{self.backward_kwargs!r}}}
            return msgpack.unpackb(inp, **kwargs)
    """)

forward(inp)

Serialize a Python object into bytes using msgpack.

Parameters:

Name Type Description Default
inp Any

Object to serialize.

 required

Returns:

Type Description
bytes

Msgpack-encoded bytes.
Source code in entry/compdata/transformation/serialization/msgpack.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def forward(self, inp: Any) -> bytes:
    """Serialize a Python object into bytes using msgpack.

    Parameters
    ----------
    inp : Any
        Object to serialize.

    Returns
    -------
    bytes
        Msgpack-encoded bytes.
    """
    kwargs = {"use_bin_type": True, **self.forward_kwargs}
    return msgpack.packb(inp, **kwargs)

backward(inp)

Deserialize bytes back into a Python object using msgpack.

Parameters:

Name Type Description Default
inp bytes

Msgpack-encoded bytes.

 required

Returns:

Type Description
Any

Deserialized Python object.
Source code in entry/compdata/transformation/serialization/msgpack.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def backward(self, inp: bytes) -> Any:
    """Deserialize bytes back into a Python object using msgpack.

    Parameters
    ----------
    inp : bytes
        Msgpack-encoded bytes.

    Returns
    -------
    Any
        Deserialized Python object.
    """
    kwargs = {"raw": False, **self.backward_kwargs}
    return msgpack.unpackb(inp, **kwargs)

NumpySerializer

Bases: _data_transformation

Reversible NumPy serialiser using np.save / np.load.

Source code in entry/compdata/transformation/serialization/numpy.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class NumpySerializer(_data_transformation):
    """Reversible NumPy serialiser using ``np.save`` / ``np.load``."""

    name: str = "numpy"

    def model_post_init(self, __context: Any) -> None:
        """Build backward_code dynamically after model creation."""
        self.backward_code = textwrap.dedent(f"""
            def backward(inp):
                import io
                import numpy as np
                buf = io.BytesIO(inp)
                kwargs = {{'allow_pickle': False, **{self.backward_kwargs!r}}}
                return np.load(buf, **kwargs)
        """)

    def forward(self, inp: np.ndarray) -> bytes:
        """Serialize a NumPy array into bytes.

        Parameters
        ----------
        inp : np.ndarray
            Array to serialize.

        Returns
        -------
        bytes
            ``np.save``-encoded bytes.
        """
        buf = io.BytesIO()
        kwargs = {"allow_pickle": False, **self.forward_kwargs}
        np.save(buf, inp, **kwargs)
        return buf.getvalue()

    def backward(self, inp: bytes) -> np.ndarray:
        """Deserialize bytes back into a NumPy array.

        Parameters
        ----------
        inp : bytes
            Bytes produced by :meth:`forward`.

        Returns
        -------
        np.ndarray
            Reconstructed array.
        """
        buf = io.BytesIO(inp)
        kwargs = {"allow_pickle": False, **self.backward_kwargs}
        return np.load(buf, **kwargs)

model_post_init(__context)

Build backward_code dynamically after model creation.

Source code in entry/compdata/transformation/serialization/numpy.py 
15
16
17
18
19
20
21
22
23
24
def model_post_init(self, __context: Any) -> None:
    """Build backward_code dynamically after model creation."""
    self.backward_code = textwrap.dedent(f"""
        def backward(inp):
            import io
            import numpy as np
            buf = io.BytesIO(inp)
            kwargs = {{'allow_pickle': False, **{self.backward_kwargs!r}}}
            return np.load(buf, **kwargs)
    """)

forward(inp)

Serialize a NumPy array into bytes.

Parameters:

Name Type Description Default
inp ndarray

Array to serialize.

 required

Returns:

Type Description
bytes

np.save-encoded bytes.
Source code in entry/compdata/transformation/serialization/numpy.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def forward(self, inp: np.ndarray) -> bytes:
    """Serialize a NumPy array into bytes.

    Parameters
    ----------
    inp : np.ndarray
        Array to serialize.

    Returns
    -------
    bytes
        ``np.save``-encoded bytes.
    """
    buf = io.BytesIO()
    kwargs = {"allow_pickle": False, **self.forward_kwargs}
    np.save(buf, inp, **kwargs)
    return buf.getvalue()

backward(inp)

Deserialize bytes back into a NumPy array.

Parameters:

Name Type Description Default
inp bytes

Bytes produced by :meth:forward.

 required

Returns:

Type Description
ndarray

Reconstructed array.
Source code in entry/compdata/transformation/serialization/numpy.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def backward(self, inp: bytes) -> np.ndarray:
    """Deserialize bytes back into a NumPy array.

    Parameters
    ----------
    inp : bytes
        Bytes produced by :meth:`forward`.

    Returns
    -------
    np.ndarray
        Reconstructed array.
    """
    buf = io.BytesIO(inp)
    kwargs = {"allow_pickle": False, **self.backward_kwargs}
    return np.load(buf, **kwargs)

TorchSerializer

Bases: _data_transformation

Reversible PyTorch serialiser using torch.save / torch.load.

Source code in entry/compdata/transformation/serialization/torch.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class TorchSerializer(_data_transformation):
    """Reversible PyTorch serialiser using ``torch.save`` / ``torch.load``."""

    name: str = "torch"

    def model_post_init(self, __context: Any) -> None:
        """Build backward_code dynamically after model creation."""
        self.backward_code = textwrap.dedent(f"""
            def backward(inp):
                import io
                import torch
                buf = io.BytesIO(inp)
                kwargs = {self.backward_kwargs!r}
                return torch.load(buf, **kwargs)
        """)

    def forward(self, inp: torch.Tensor) -> bytes:
        """Serialize a PyTorch tensor into bytes.

        Parameters
        ----------
        inp : torch.Tensor
            Tensor to serialize.

        Returns
        -------
        bytes
            ``torch.save``-encoded bytes.
        """
        buf = io.BytesIO()
        kwargs = {**self.forward_kwargs}
        torch.save(inp, buf, **kwargs)
        return buf.getvalue()

    def backward(self, inp: bytes) -> Any:
        """Deserialize bytes back into a PyTorch tensor.

        Parameters
        ----------
        inp : bytes
            Bytes produced by :meth:`forward`.

        Returns
        -------
        Any
            Reconstructed tensor.
        """
        buf = io.BytesIO(inp)
        kwargs = {**self.backward_kwargs}
        return torch.load(buf, **kwargs)

model_post_init(__context)

Build backward_code dynamically after model creation.

Source code in entry/compdata/transformation/serialization/torch.py 
15
16
17
18
19
20
21
22
23
24
def model_post_init(self, __context: Any) -> None:
    """Build backward_code dynamically after model creation."""
    self.backward_code = textwrap.dedent(f"""
        def backward(inp):
            import io
            import torch
            buf = io.BytesIO(inp)
            kwargs = {self.backward_kwargs!r}
            return torch.load(buf, **kwargs)
    """)

forward(inp)

Serialize a PyTorch tensor into bytes.

Parameters:

Name Type Description Default
inp Tensor

Tensor to serialize.

 required

Returns:

Type Description
bytes

torch.save-encoded bytes.
Source code in entry/compdata/transformation/serialization/torch.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def forward(self, inp: torch.Tensor) -> bytes:
    """Serialize a PyTorch tensor into bytes.

    Parameters
    ----------
    inp : torch.Tensor
        Tensor to serialize.

    Returns
    -------
    bytes
        ``torch.save``-encoded bytes.
    """
    buf = io.BytesIO()
    kwargs = {**self.forward_kwargs}
    torch.save(inp, buf, **kwargs)
    return buf.getvalue()

backward(inp)

Deserialize bytes back into a PyTorch tensor.

Parameters:

Name Type Description Default
inp bytes

Bytes produced by :meth:forward.

 required

Returns:

Type Description
Any

Reconstructed tensor.
Source code in entry/compdata/transformation/serialization/torch.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def backward(self, inp: bytes) -> Any:
    """Deserialize bytes back into a PyTorch tensor.

    Parameters
    ----------
    inp : bytes
        Bytes produced by :meth:`forward`.

    Returns
    -------
    Any
        Reconstructed tensor.
    """
    buf = io.BytesIO(inp)
    kwargs = {**self.backward_kwargs}
    return torch.load(buf, **kwargs)

JsonString

Bases: _data_transformation

Reversible JSON string transformation.

Forward serialises a Python object to a compact JSON string; backward parses it back.

Source code in entry/compdata/transformation/jsonstring/jsonstring.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class JsonString(_data_transformation):
    """Reversible JSON string transformation.

    Forward serialises a Python object to a compact JSON string;
    backward parses it back.
    """

    name: str = "json_string"

    def model_post_init(self, __context: Any) -> None:
        """Build standalone backward recovery code."""
        # Standalone recovery code mirroring `backward()` with embedded kwargs
        self.backward_code = textwrap.dedent(f"""
            def backward(data):
                import json
                if not isinstance(data, str):
                    raise TypeError("JsonString.backward expects a JSON string (str)")
                return json.loads(data)
        """)


    def forward(self, data: Any) -> str:
        """Serialize *data* to a compact JSON string.

        Parameters
        ----------
        data : Any
            JSON-serialisable Python object.

        Returns
        -------
        str
            Compact JSON string.

        Raises
        ------
        TypeError
            If *data* is not JSON-serialisable.
        """
        try:
            return json.dumps(
                data,
                separators=(",", ":"),
                ensure_ascii=False,
                **self.forward_kwargs,
            )
        except (TypeError, ValueError) as e:
            raise TypeError(f"JsonString.forward: object not JSON serializable: {e!s}")


    def backward(self, data: str) -> Any:
        """Deserialize a JSON string back to a Python object.

        Parameters
        ----------
        data : str
            JSON string.

        Returns
        -------
        Any
            Parsed Python object.

        Raises
        ------
        TypeError
            If *data* is not a string or contains invalid JSON.
        """
        if not isinstance(data, str):
            raise TypeError("JsonString.backward expects a JSON string (str)")
        try:
            return json.loads(data, **self.backward_kwargs)
        except json.JSONDecodeError as e:
            raise TypeError(f"JsonString.backward: invalid JSON string: {e.msg}")

model_post_init(__context)

Build standalone backward recovery code.

Source code in entry/compdata/transformation/jsonstring/jsonstring.py 
18
19
20
21
22
23
24
25
26
27
def model_post_init(self, __context: Any) -> None:
    """Build standalone backward recovery code."""
    # Standalone recovery code mirroring `backward()` with embedded kwargs
    self.backward_code = textwrap.dedent(f"""
        def backward(data):
            import json
            if not isinstance(data, str):
                raise TypeError("JsonString.backward expects a JSON string (str)")
            return json.loads(data)
    """)

forward(data)

Serialize data to a compact JSON string.

Parameters:

Name Type Description Default
data Any

JSON-serialisable Python object.

 required

Returns:

Type Description
str

Compact JSON string.

Raises:

Type Description
TypeError

If data is not JSON-serialisable.
Source code in entry/compdata/transformation/jsonstring/jsonstring.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def forward(self, data: Any) -> str:
    """Serialize *data* to a compact JSON string.

    Parameters
    ----------
    data : Any
        JSON-serialisable Python object.

    Returns
    -------
    str
        Compact JSON string.

    Raises
    ------
    TypeError
        If *data* is not JSON-serialisable.
    """
    try:
        return json.dumps(
            data,
            separators=(",", ":"),
            ensure_ascii=False,
            **self.forward_kwargs,
        )
    except (TypeError, ValueError) as e:
        raise TypeError(f"JsonString.forward: object not JSON serializable: {e!s}")

backward(data)

Deserialize a JSON string back to a Python object.

Parameters:

Name Type Description Default
data str

JSON string.

 required

Returns:

Type Description
Any

Parsed Python object.

Raises:

Type Description
TypeError

If data is not a string or contains invalid JSON.
Source code in entry/compdata/transformation/jsonstring/jsonstring.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def backward(self, data: str) -> Any:
    """Deserialize a JSON string back to a Python object.

    Parameters
    ----------
    data : str
        JSON string.

    Returns
    -------
    Any
        Parsed Python object.

    Raises
    ------
    TypeError
        If *data* is not a string or contains invalid JSON.
    """
    if not isinstance(data, str):
        raise TypeError("JsonString.backward expects a JSON string (str)")
    try:
        return json.loads(data, **self.backward_kwargs)
    except json.JSONDecodeError as e:
        raise TypeError(f"JsonString.backward: invalid JSON string: {e.msg}")