omnipy.components.json.serializers

Tar-file serializer for Omnipy JSON datasets.

CLASS	DESCRIPTION
JsonDatasetToTarFileSerializer	Serialize JSON datasets to and from gzipped tar archives.

JsonDatasetToTarFileSerializer

Bases: TarFileSerializer[JsonBaseDataset]

              flowchart BT
              omnipy.components.json.serializers.JsonDatasetToTarFileSerializer[JsonDatasetToTarFileSerializer]
              omnipy.data.serializer.TarFileSerializer[TarFileSerializer]
              omnipy.data.serializer.Serializer[Serializer]

                              omnipy.data.serializer.TarFileSerializer --> omnipy.components.json.serializers.JsonDatasetToTarFileSerializer
                                omnipy.data.serializer.Serializer --> omnipy.data.serializer.TarFileSerializer
                



              click omnipy.components.json.serializers.JsonDatasetToTarFileSerializer href "" "omnipy.components.json.serializers.JsonDatasetToTarFileSerializer"
              click omnipy.data.serializer.TarFileSerializer href "" "omnipy.data.serializer.TarFileSerializer"
              click omnipy.data.serializer.Serializer href "" "omnipy.data.serializer.Serializer"
            

Serialize JSON datasets to and from gzipped tar archives.

METHOD	DESCRIPTION
deserialize	Deserialize a gzipped tar archive back into a dataset.
get_dataset_cls_for_new	Return the dataset class created during deserialization.
get_output_file_suffix	Return the file suffix used for serialized dataset members.
is_dataset_directly_supported	Return whether a dataset can be serialized as JSON files.
serialize	Serialize a dataset into a gzipped tar archive.

Source code in src/omnipy/components/json/serializers.py

class JsonDatasetToTarFileSerializer(TarFileSerializer[JsonBaseDataset]):
    """Serialize JSON datasets to and from gzipped tar archives."""
    @classmethod
    def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
        """Return whether a dataset can be serialized as JSON files."""

        from ..isa.datasets import IsaJsonDataset
        return isinstance(dataset, JsonBaseDataset) or isinstance(dataset, IsaJsonDataset)

    @classmethod
    def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{SERIALIZER_GET_DATASET_CLS_FOR_NEW_SUMMARY}}
        """Return the dataset class created during deserialization."""

        return JsonDataset

    @classmethod
    def get_output_file_suffix(cls) -> str:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{SERIALIZER_GET_OUTPUT_FILE_SUFFIX_SUMMARY}}
        """Return the file suffix used for serialized dataset members."""

        return 'json'

    @classmethod
    def serialize(cls, dataset: JsonBaseDataset) -> bytes | memoryview:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{SERIALIZE_GZIPPED_TAR_SUMMARY}}
        """Serialize a dataset into a gzipped tar archive."""
        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:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{DESERIALIZE_GZIPPED_TAR_SUMMARY}}
        """Deserialize a gzipped tar archive back into a dataset."""

        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

Deserialize a gzipped tar archive back into a dataset.

Source code in src/omnipy/components/json/serializers.py

@classmethod
def deserialize(cls, serialized: bytes, any_file_suffix=False) -> JsonDataset:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{DESERIALIZE_GZIPPED_TAR_SUMMARY}}
    """Deserialize a gzipped tar archive back into a dataset."""

    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]

Return the dataset class created during deserialization.

Source code in src/omnipy/components/json/serializers.py

@classmethod
def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{SERIALIZER_GET_DATASET_CLS_FOR_NEW_SUMMARY}}
    """Return the dataset class created during deserialization."""

    return JsonDataset

get_output_file_suffix classmethod

get_output_file_suffix() -> str

Return the file suffix used for serialized dataset members.

Source code in src/omnipy/components/json/serializers.py

@classmethod
def get_output_file_suffix(cls) -> str:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{SERIALIZER_GET_OUTPUT_FILE_SUFFIX_SUMMARY}}
    """Return the file suffix used for serialized dataset members."""

    return 'json'

is_dataset_directly_supported classmethod

is_dataset_directly_supported(dataset: IsDataset) -> bool

Return whether a dataset can be serialized as JSON files.

Source code in src/omnipy/components/json/serializers.py

@classmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    """Return whether a dataset can be serialized as JSON files."""

    from ..isa.datasets import IsaJsonDataset
    return isinstance(dataset, JsonBaseDataset) or isinstance(dataset, IsaJsonDataset)

serialize classmethod

serialize(dataset: JsonBaseDataset) -> bytes | memoryview

Serialize a dataset into a gzipped tar archive.

Source code in src/omnipy/components/json/serializers.py

@classmethod
def serialize(cls, dataset: JsonBaseDataset) -> bytes | memoryview:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{SERIALIZE_GZIPPED_TAR_SUMMARY}}
    """Serialize a dataset into a gzipped tar archive."""
    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)