Skip to content

serialization

Attributes

Classes

SerializeConfig

Bases: KiaraModuleConfig

Source code in kiara/modules/included_core_modules/serialization.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class SerializeConfig(KiaraModuleConfig):

    value_type: str = Field(
        description="The value type of the actual (unserialized) value."
    )
    target_profile: str = Field(
        description="The profile name of the de-serialization result data."
    )
    serialization_profile: str = Field(
        description="The name of the serialization profile used to serialize the source value."
    )

    @validator("value_type")
    def validate_source_type(cls, value):
        if value == "serialization_config":
            raise ValueError(f"Invalid source type: {value}.")
        return value

Attributes

value_type: str = Field(description='The value type of the actual (unserialized) value.') class-attribute
target_profile: str = Field(description='The profile name of the de-serialization result data.') class-attribute
serialization_profile: str = Field(description='The name of the serialization profile used to serialize the source value.') class-attribute

Functions

validate_source_type(value)
Source code in kiara/modules/included_core_modules/serialization.py 
39
40
41
42
43
@validator("value_type")
def validate_source_type(cls, value):
    if value == "serialization_config":
        raise ValueError(f"Invalid source type: {value}.")
    return value

DeserializeValueModule

Bases: KiaraModule

Source code in kiara/modules/included_core_modules/serialization.py 
 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
class DeserializeValueModule(KiaraModule):

    _config_cls = SerializeConfig

    @classmethod
    @abc.abstractmethod
    def retrieve_serialized_value_type(cls) -> str:
        raise NotImplementedError()

    @classmethod
    @abc.abstractmethod
    def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
        raise NotImplementedError()

    @classmethod
    @abc.abstractmethod
    def retrieve_supported_serialization_profile(cls) -> str:
        raise NotImplementedError()

    def create_inputs_schema(
        self,
    ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

        value_type = self.get_config_value("value_type")
        return {
            value_type: {
                "type": value_type,
                "doc": "The value object.",
            },
            "deserialization_config": {
                "type": "any",
                "doc": "Serialization-format specific configuration.",
                "optional": True,
            },
        }

    def create_outputs_schema(
        self,
    ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

        return {
            "python_object": {
                "type": "python_object",
                "doc": "The deserialized python object instance.",
            },
        }

    def _retrieve_module_characteristics(self) -> ModuleCharacteristics:
        return DEFAULT_IDEMPOTENT_INTERNAL_MODULE_CHARACTERISTICS

    def process(self, inputs: ValueMap, outputs: ValueMap) -> None:

        value_type = self.get_config_value("value_type")
        serialized_value = inputs.get_value_obj(value_type)
        config = inputs.get_value_obj("deserialization_config")

        target_profile = self.get_config_value("target_profile")
        func_name = f"to__{target_profile}"
        func = getattr(self, func_name)

        if config.is_set:
            _config = config.data
        else:
            _config = {}

        result: Any = func(data=serialized_value.serialized_data, **_config)
        outputs.set_value("python_object", result)

Attributes

_config_cls = SerializeConfig class-attribute

Functions

retrieve_serialized_value_type() -> str abstractmethod classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
50
51
52
53
@classmethod
@abc.abstractmethod
def retrieve_serialized_value_type(cls) -> str:
    raise NotImplementedError()
retrieve_supported_target_profiles() -> Mapping[str, Type] abstractmethod classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
55
56
57
58
@classmethod
@abc.abstractmethod
def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
    raise NotImplementedError()
retrieve_supported_serialization_profile() -> str abstractmethod classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
60
61
62
63
@classmethod
@abc.abstractmethod
def retrieve_supported_serialization_profile(cls) -> str:
    raise NotImplementedError()
create_inputs_schema() -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]
Source code in kiara/modules/included_core_modules/serialization.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def create_inputs_schema(
    self,
) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

    value_type = self.get_config_value("value_type")
    return {
        value_type: {
            "type": value_type,
            "doc": "The value object.",
        },
        "deserialization_config": {
            "type": "any",
            "doc": "Serialization-format specific configuration.",
            "optional": True,
        },
    }
create_outputs_schema() -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]
Source code in kiara/modules/included_core_modules/serialization.py 
82
83
84
85
86
87
88
89
90
91
def create_outputs_schema(
    self,
) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

    return {
        "python_object": {
            "type": "python_object",
            "doc": "The deserialized python object instance.",
        },
    }
process(inputs: ValueMap, outputs: ValueMap) -> None
Source code in kiara/modules/included_core_modules/serialization.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def process(self, inputs: ValueMap, outputs: ValueMap) -> None:

    value_type = self.get_config_value("value_type")
    serialized_value = inputs.get_value_obj(value_type)
    config = inputs.get_value_obj("deserialization_config")

    target_profile = self.get_config_value("target_profile")
    func_name = f"to__{target_profile}"
    func = getattr(self, func_name)

    if config.is_set:
        _config = config.data
    else:
        _config = {}

    result: Any = func(data=serialized_value.serialized_data, **_config)
    outputs.set_value("python_object", result)

UnpickleModule

Bases: DeserializeValueModule

Source code in kiara/modules/included_core_modules/serialization.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class UnpickleModule(DeserializeValueModule):

    _module_type_name = "unpickle.value"

    @classmethod
    def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:

        return {"python_object": object}

    @classmethod
    def retrieve_supported_serialization_profile(cls) -> str:
        return "pickle"

    @classmethod
    def retrieve_serialized_value_type(cls) -> str:

        return "any"

    def to__python_object(self, data: SerializedData, **config: Any):

        try:
            import pickle5 as pickle
        except Exception:
            import pickle  # type: ignore

        assert "python_object" in data.get_keys()
        python_object_data = data.get_serialized_data("python_object")
        assert python_object_data.get_number_of_chunks() == 1

        _bytes = list(python_object_data.get_chunks(as_files=False))[0]
        data = pickle.loads(_bytes)

        return data

Functions

retrieve_supported_target_profiles() -> Mapping[str, Type] classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
119
120
121
122
@classmethod
def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:

    return {"python_object": object}
retrieve_supported_serialization_profile() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
124
125
126
@classmethod
def retrieve_supported_serialization_profile(cls) -> str:
    return "pickle"
retrieve_serialized_value_type() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
128
129
130
131
@classmethod
def retrieve_serialized_value_type(cls) -> str:

    return "any"
to__python_object(data: SerializedData, **config: Any)
Source code in kiara/modules/included_core_modules/serialization.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def to__python_object(self, data: SerializedData, **config: Any):

    try:
        import pickle5 as pickle
    except Exception:
        import pickle  # type: ignore

    assert "python_object" in data.get_keys()
    python_object_data = data.get_serialized_data("python_object")
    assert python_object_data.get_number_of_chunks() == 1

    _bytes = list(python_object_data.get_chunks(as_files=False))[0]
    data = pickle.loads(_bytes)

    return data

LoadBytesModule

Bases: DeserializeValueModule

Source code in kiara/modules/included_core_modules/serialization.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class LoadBytesModule(DeserializeValueModule):

    _module_type_name = "load.bytes"

    @classmethod
    def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
        return {"python_object": bytes}

    @classmethod
    def retrieve_supported_serialization_profile(cls) -> str:
        return "raw"

    @classmethod
    def retrieve_serialized_value_type(cls) -> str:
        return "bytes"

    def to__python_object(self, data: SerializedData, **config: Any) -> bytes:

        chunks = data.get_serialized_data("bytes")
        assert chunks.get_number_of_chunks() == 1
        _chunks = list(chunks.get_chunks(as_files=False))
        assert len(_chunks) == 1
        _chunk: bytes = _chunks[0]  # type: ignore
        return _chunk

Functions

retrieve_supported_target_profiles() -> Mapping[str, Type] classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
154
155
156
@classmethod
def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
    return {"python_object": bytes}
retrieve_supported_serialization_profile() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
158
159
160
@classmethod
def retrieve_supported_serialization_profile(cls) -> str:
    return "raw"
retrieve_serialized_value_type() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
162
163
164
@classmethod
def retrieve_serialized_value_type(cls) -> str:
    return "bytes"
to__python_object(data: SerializedData, **config: Any) -> bytes
Source code in kiara/modules/included_core_modules/serialization.py 
166
167
168
169
170
171
172
173
def to__python_object(self, data: SerializedData, **config: Any) -> bytes:

    chunks = data.get_serialized_data("bytes")
    assert chunks.get_number_of_chunks() == 1
    _chunks = list(chunks.get_chunks(as_files=False))
    assert len(_chunks) == 1
    _chunk: bytes = _chunks[0]  # type: ignore
    return _chunk

LoadStringModule

Bases: DeserializeValueModule

Source code in kiara/modules/included_core_modules/serialization.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class LoadStringModule(DeserializeValueModule):

    _module_type_name = "load.string"

    @classmethod
    def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
        return {"python_object": str}

    @classmethod
    def retrieve_supported_serialization_profile(cls) -> str:
        return "raw"

    @classmethod
    def retrieve_serialized_value_type(cls) -> str:
        return "string"

    def to__python_object(self, data: SerializedData, **config: Any) -> str:

        chunks = data.get_serialized_data("string")
        assert chunks.get_number_of_chunks() == 1
        _chunks = list(chunks.get_chunks(as_files=False))
        assert len(_chunks) == 1

        bytes_string: bytes = _chunks[0]  # type: ignore
        return bytes_string.decode("utf-8")

Functions

retrieve_supported_target_profiles() -> Mapping[str, Type] classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
180
181
182
@classmethod
def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
    return {"python_object": str}
retrieve_supported_serialization_profile() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
184
185
186
@classmethod
def retrieve_supported_serialization_profile(cls) -> str:
    return "raw"
retrieve_serialized_value_type() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
188
189
190
@classmethod
def retrieve_serialized_value_type(cls) -> str:
    return "string"
to__python_object(data: SerializedData, **config: Any) -> str
Source code in kiara/modules/included_core_modules/serialization.py 
192
193
194
195
196
197
198
199
200
def to__python_object(self, data: SerializedData, **config: Any) -> str:

    chunks = data.get_serialized_data("string")
    assert chunks.get_number_of_chunks() == 1
    _chunks = list(chunks.get_chunks(as_files=False))
    assert len(_chunks) == 1

    bytes_string: bytes = _chunks[0]  # type: ignore
    return bytes_string.decode("utf-8")

LoadInternalModel

Bases: DeserializeValueModule

Source code in kiara/modules/included_core_modules/serialization.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class LoadInternalModel(DeserializeValueModule):

    _module_type_name = "load.internal_model"

    @classmethod
    def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
        return {"python_object": KiaraModel}

    @classmethod
    def retrieve_supported_serialization_profile(cls) -> str:
        return "json"

    @classmethod
    def retrieve_serialized_value_type(cls) -> str:
        return "internal_model"

    def to__python_object(self, data: SerializedData, **config: Any) -> KiaraModel:

        chunks = data.get_serialized_data("data")
        assert chunks.get_number_of_chunks() == 1
        _chunks = list(chunks.get_chunks(as_files=False))
        assert len(_chunks) == 1

        bytes_string: bytes = _chunks[0]  # type: ignore
        model_data = orjson.loads(bytes_string)

        model_id: str = data.data_type_config["kiara_model_id"]
        model_registry = ModelRegistry.instance()
        m_cls = model_registry.get_model_cls(kiara_model_id=model_id)
        obj = m_cls(**model_data)
        return obj

Functions

retrieve_supported_target_profiles() -> Mapping[str, Type] classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
207
208
209
@classmethod
def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
    return {"python_object": KiaraModel}
retrieve_supported_serialization_profile() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
211
212
213
@classmethod
def retrieve_supported_serialization_profile(cls) -> str:
    return "json"
retrieve_serialized_value_type() -> str classmethod
Source code in kiara/modules/included_core_modules/serialization.py 
215
216
217
@classmethod
def retrieve_serialized_value_type(cls) -> str:
    return "internal_model"
to__python_object(data: SerializedData, **config: Any) -> KiaraModel
Source code in kiara/modules/included_core_modules/serialization.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def to__python_object(self, data: SerializedData, **config: Any) -> KiaraModel:

    chunks = data.get_serialized_data("data")
    assert chunks.get_number_of_chunks() == 1
    _chunks = list(chunks.get_chunks(as_files=False))
    assert len(_chunks) == 1

    bytes_string: bytes = _chunks[0]  # type: ignore
    model_data = orjson.loads(bytes_string)

    model_id: str = data.data_type_config["kiara_model_id"]
    model_registry = ModelRegistry.instance()
    m_cls = model_registry.get_model_cls(kiara_model_id=model_id)
    obj = m_cls(**model_data)
    return obj

DeserializeJsonConfig

Bases: KiaraModuleConfig

Source code in kiara/modules/included_core_modules/serialization.py 
236
237
238
239
240
class DeserializeJsonConfig(KiaraModuleConfig):

    result_path: Union[str, None] = Field(
        description="The path of the result dictionary to return.", default="data"
    )

Attributes

result_path: Union[str, None] = Field(description='The path of the result dictionary to return.', default='data') class-attribute

DeserializeFromJsonModule

Bases: KiaraModule

Source code in kiara/modules/included_core_modules/serialization.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
class DeserializeFromJsonModule(KiaraModule):

    _module_type_name: str = "deserialize.from_json"
    _config_cls = DeserializeJsonConfig

    def _retrieve_module_characteristics(self) -> ModuleCharacteristics:
        return DEFAULT_IDEMPOTENT_INTERNAL_MODULE_CHARACTERISTICS

    def create_inputs_schema(
        self,
    ) -> ValueMapSchema:

        return {
            "value": {
                "type": "any",
                "doc": "The value object to deserialize the data for.",
            }
        }

    def create_outputs_schema(
        self,
    ) -> ValueMapSchema:

        return {
            "python_object": {
                "type": "python_object",
                "doc": "The deserialized python object.",
            }
        }

    def process(self, inputs: ValueMap, outputs: ValueMap):

        value: Value = inputs.get_value_obj("value")
        serialized: SerializedData = value.serialized_data

        chunks = serialized.get_serialized_data(self.get_config_value("result_path"))
        assert chunks.get_number_of_chunks() == 1
        _data = list(chunks.get_chunks(as_files=False))
        assert len(_data) == 1
        _chunk: bytes = _data[0]  # type: ignore

        deserialized = orjson.loads(_chunk)

        outputs.set_value("python_object", deserialized)

Attributes

_config_cls = DeserializeJsonConfig class-attribute

Functions

create_inputs_schema() -> ValueMapSchema
Source code in kiara/modules/included_core_modules/serialization.py 
251
252
253
254
255
256
257
258
259
260
def create_inputs_schema(
    self,
) -> ValueMapSchema:

    return {
        "value": {
            "type": "any",
            "doc": "The value object to deserialize the data for.",
        }
    }
create_outputs_schema() -> ValueMapSchema
Source code in kiara/modules/included_core_modules/serialization.py 
262
263
264
265
266
267
268
269
270
271
def create_outputs_schema(
    self,
) -> ValueMapSchema:

    return {
        "python_object": {
            "type": "python_object",
            "doc": "The deserialized python object.",
        }
    }
process(inputs: ValueMap, outputs: ValueMap)
Source code in kiara/modules/included_core_modules/serialization.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def process(self, inputs: ValueMap, outputs: ValueMap):

    value: Value = inputs.get_value_obj("value")
    serialized: SerializedData = value.serialized_data

    chunks = serialized.get_serialized_data(self.get_config_value("result_path"))
    assert chunks.get_number_of_chunks() == 1
    _data = list(chunks.get_chunks(as_files=False))
    assert len(_data) == 1
    _chunk: bytes = _data[0]  # type: ignore

    deserialized = orjson.loads(_chunk)

    outputs.set_value("python_object", deserialized)