pyany-serde 0.6.0

Serialization and deserialization for Python objects
Documentation 
1
2
3
4
5
6
7
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
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

import pickle
from typing import Any, Literal

import numpy as np
from pyany_serde import NumpySerdeConfig
from pyany_serde.pydantic_numpy_serde_config_tests import (  # pyright:  ignore [reportMissingImports]
    validate_eq,  # pyright: ignore [reportUnknownVariableType]
)

from pydantic import BaseModel


class MyModel(BaseModel):
    my_field: NumpySerdeConfig


class MyClass:
    val: int
    val2: int

    def __init__(self, val: int, val2: int):
        self.val = val
        self.val2 = val2


def preprocessor_fn(v: MyClass):
    return np.array([v.val, v.val2], dtype=np.int64)


def postprocessor_fn(v: np.ndarray[tuple[Literal[2]], np.dtype[np.int64]]):
    return MyClass(v[0], v[1])


def validate_model_construction_flows(
    expected: NumpySerdeConfig, my_field_dict: dict[str, Any], model_json: str
):
    # constructor with instance
    actual = MyModel(my_field=expected).my_field
    validate_eq(expected, actual, "$")

    # constructor with dict
    actual = MyModel(my_field=my_field_dict).my_field  # pyright: ignore [reportArgumentType]
    validate_eq(expected, actual, "$")

    # validate json
    actual = MyModel.model_validate_json(model_json).my_field
    validate_eq(expected, actual, "$")

    # validate instance
    actual = MyModel.model_validate(MyModel(my_field=expected)).my_field
    validate_eq(expected, actual, "$")

    # validate dict
    actual = MyModel.model_validate({"my_field": my_field_dict}).my_field
    validate_eq(expected, actual, "$")


def test_dynamic():
    expected = NumpySerdeConfig.DYNAMIC(
        preprocessor_fn=preprocessor_fn, postprocessor_fn=postprocessor_fn
    )
    my_field_dict = {
        "type": "dynamic",
        "preprocessor_fn_pkl": pickle.dumps(preprocessor_fn).hex(),
        "postprocessor_fn_pkl": pickle.dumps(postprocessor_fn).hex(),
    }
    model_json = f"""
{{
    "my_field": {{
        "type": "dynamic",
        "preprocessor_fn_pkl": "{pickle.dumps(preprocessor_fn).hex()}",
        "postprocessor_fn_pkl": "{pickle.dumps(postprocessor_fn).hex()}"
    }}
}}
"""
    validate_model_construction_flows(expected, my_field_dict, model_json)


def test_static():
    expected = NumpySerdeConfig.STATIC(
        shape=(2,),
        preprocessor_fn=preprocessor_fn,
        postprocessor_fn=postprocessor_fn,
        allocation_pool_min_size=0,
        allocation_pool_max_size=10,
        allocation_pool_warning_size=1,
    )
    my_field_dict = {
        "type": "static",
        "shape": [2],
        "preprocessor_fn_pkl": pickle.dumps(preprocessor_fn).hex(),
        "postprocessor_fn_pkl": pickle.dumps(postprocessor_fn).hex(),
        "allocation_pool_min_size": 0,
        "allocation_pool_max_size": 10,
        "allocation_pool_warning_size": 1,
    }
    model_json = f"""
{{
    "my_field": {{
        "type": "static",
        "shape": [2],
        "preprocessor_fn_pkl": "{pickle.dumps(preprocessor_fn).hex()}",
        "postprocessor_fn_pkl": "{pickle.dumps(postprocessor_fn).hex()}",
        "allocation_pool_min_size": 0,
        "allocation_pool_max_size": 10,
        "allocation_pool_warning_size": 1
    }}
}}
"""
    validate_model_construction_flows(expected, my_field_dict, model_json)