Skip to content

omnipy.components.json.serializers

CLASS DESCRIPTION
JsonDatasetToTarFileSerializer

JsonDatasetToTarFileSerializer

Bases: TarFileSerializer[JsonBaseDataset]

METHOD DESCRIPTION
deserialize
get_dataset_cls_for_new
get_output_file_suffix
is_dataset_directly_supported
serialize
Source code in src/omnipy/components/json/serializers.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
class JsonDatasetToTarFileSerializer(TarFileSerializer[JsonBaseDataset]):
    """"""
    @classmethod
    def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
        from ..isa.datasets import IsaJsonDataset
        return isinstance(dataset, JsonBaseDataset) or isinstance(dataset, IsaJsonDataset)

    @classmethod
    def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
        return JsonDataset

    @classmethod
    def get_output_file_suffix(cls) -> str:
        return 'json'

    @classmethod
    def serialize(cls, dataset: JsonBaseDataset) -> bytes | memoryview:
        def json_encode_func(json_data: JsonModel) -> bytes:
            return json_data.to_json().encode('utf8')

        return cls.create_tarfile_from_dataset(dataset, data_encode_func=json_encode_func)

    @classmethod
    def deserialize(cls, serialized: bytes, any_file_suffix=False) -> JsonDataset:
        json_dataset = JsonDataset()

        def json_decode_func(file_stream: IO[bytes]) -> str:
            return file_stream.read().decode('utf8')

        def json_dictify_object(data_file: str, obj_val: str) -> dict[str, str]:
            return {f'{data_file}': f'{obj_val}'}

        cls.create_dataset_from_tarfile(
            json_dataset,
            serialized,
            data_decode_func=json_decode_func,
            dictify_object_func=json_dictify_object,
            import_method='from_json',
            any_file_suffix=any_file_suffix,
        )

        return json_dataset

deserialize classmethod 

deserialize(serialized: bytes, any_file_suffix=False) -> JsonDataset
Source code in src/omnipy/components/json/serializers.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@classmethod
def deserialize(cls, serialized: bytes, any_file_suffix=False) -> JsonDataset:
    json_dataset = JsonDataset()

    def json_decode_func(file_stream: IO[bytes]) -> str:
        return file_stream.read().decode('utf8')

    def json_dictify_object(data_file: str, obj_val: str) -> dict[str, str]:
        return {f'{data_file}': f'{obj_val}'}

    cls.create_dataset_from_tarfile(
        json_dataset,
        serialized,
        data_decode_func=json_decode_func,
        dictify_object_func=json_dictify_object,
        import_method='from_json',
        any_file_suffix=any_file_suffix,
    )

    return json_dataset

get_dataset_cls_for_new classmethod 

get_dataset_cls_for_new() -> Type[IsDataset]
Source code in src/omnipy/components/json/serializers.py 
17
18
19
@classmethod
def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
    return JsonDataset

get_output_file_suffix classmethod 

get_output_file_suffix() -> str
Source code in src/omnipy/components/json/serializers.py 
21
22
23
@classmethod
def get_output_file_suffix(cls) -> str:
    return 'json'

is_dataset_directly_supported classmethod 

is_dataset_directly_supported(dataset: IsDataset) -> bool
Source code in src/omnipy/components/json/serializers.py 
12
13
14
15
@classmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    from ..isa.datasets import IsaJsonDataset
    return isinstance(dataset, JsonBaseDataset) or isinstance(dataset, IsaJsonDataset)

serialize classmethod 

serialize(dataset: JsonBaseDataset) -> bytes | memoryview
Source code in src/omnipy/components/json/serializers.py 
25
26
27
28
29
30
@classmethod
def serialize(cls, dataset: JsonBaseDataset) -> bytes | memoryview:
    def json_encode_func(json_data: JsonModel) -> bytes:
        return json_data.to_json().encode('utf8')

    return cls.create_tarfile_from_dataset(dataset, data_encode_func=json_encode_func)