omnipy.data.serializer

Serializer abstractions and registry helpers for Omnipy datasets.

CLASS	DESCRIPTION
Serializer	Abstract base class for dataset serializers used by Omnipy import/export flows.
SerializerRegistry	Registry and auto-detection helper for the serializers available to Omnipy.
TarFileSerializer	Serializer base class for datasets stored as gzipped tar archives of item files.

Serializer

Bases: ABC, Generic[_DatasetT]

              flowchart BT
              omnipy.data.serializer.Serializer[Serializer]

              

              click omnipy.data.serializer.Serializer href "" "omnipy.data.serializer.Serializer"
            

Abstract base class for dataset serializers used by Omnipy import/export flows.

Subclassed by:

• Reference Code reference omnipy data  serializer  TarFileSerializer

METHOD	DESCRIPTION
deserialize	Deserialize serialized bytes into a dataset instance.
get_dataset_cls_for_new	Return the dataset class created when deserializing with this serializer.
get_output_file_suffix	Return the filename suffix Omnipy associates with this serializer's output.
is_dataset_directly_supported	Return whether dataset can be reconstructed directly by this serializer.
serialize	Serialize dataset into bytes suitable for persistence or transport.

Source code in src/omnipy/data/serializer.py

class Serializer(ABC, Generic[_DatasetT]):
    """Abstract base class for dataset serializers used by Omnipy import/export flows."""
    @classmethod
    @abstractmethod
    def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
        """Return whether ``dataset`` can be reconstructed directly by this serializer."""

        pass

    @classmethod
    @abstractmethod
    def get_dataset_cls_for_new(cls) -> type[IsDataset]:
        """Return the dataset class created when deserializing with this serializer."""

        pass

    @classmethod
    @abstractmethod
    def get_output_file_suffix(cls) -> str:
        """Return the filename suffix Omnipy associates with this serializer's output."""

        pass

    @classmethod
    @abstractmethod
    def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
        """Serialize ``dataset`` into bytes suitable for persistence or transport."""

        pass

    @classmethod
    @abstractmethod
    def deserialize(cls, serialized: bytes, any_file_suffix=False) -> _DatasetT:
        """Deserialize ``serialized`` bytes into a dataset instance."""

        pass

deserialize abstractmethod classmethod

deserialize(serialized: bytes, any_file_suffix=False) -> _DatasetT

Deserialize serialized bytes into a dataset instance.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def deserialize(cls, serialized: bytes, any_file_suffix=False) -> _DatasetT:
    """Deserialize ``serialized`` bytes into a dataset instance."""

    pass

get_dataset_cls_for_new abstractmethod classmethod

get_dataset_cls_for_new() -> type[IsDataset]

Return the dataset class created when deserializing with this serializer.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def get_dataset_cls_for_new(cls) -> type[IsDataset]:
    """Return the dataset class created when deserializing with this serializer."""

    pass

get_output_file_suffix abstractmethod classmethod

get_output_file_suffix() -> str

Return the filename suffix Omnipy associates with this serializer's output.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def get_output_file_suffix(cls) -> str:
    """Return the filename suffix Omnipy associates with this serializer's output."""

    pass

is_dataset_directly_supported abstractmethod classmethod

is_dataset_directly_supported(dataset: IsDataset) -> bool

Return whether dataset can be reconstructed directly by this serializer.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    """Return whether ``dataset`` can be reconstructed directly by this serializer."""

    pass

serialize abstractmethod classmethod

serialize(dataset: _DatasetT) -> bytes | memoryview

Serialize dataset into bytes suitable for persistence or transport.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
    """Serialize ``dataset`` into bytes suitable for persistence or transport."""

    pass

SerializerRegistry

Registry and auto-detection helper for the serializers available to Omnipy.

Used by:

• Reference Code reference omnipy hub  runtime  RuntimeObjects  serializers

METHOD	DESCRIPTION
__init__
auto_detect	Return the first compatible dataset/serializer pair from the registry.
auto_detect_tar_file_serializer	Try only tar-file serializers and return the first compatible pair.
detect_tar_file_serializers_from_dataset_cls	Return tar-file serializers that can load data into dataset.
detect_tar_file_serializers_from_file_suffix	Return tar-file serializers whose output suffix matches file_suffix.
load_from_tar_file_path_based_on_dataset_cls	Load a tar archive by trying serializers compatible with to_dataset.
load_from_tar_file_path_based_on_file_suffix	Load a tar archive by detecting its serializer from member file suffixes.
register	Register a serializer class for later lookup and auto-detection.

ATTRIBUTE	DESCRIPTION
serializers	Return all registered serializer classes in registration order. TYPE: tuple[Type[IsSerializer], ...]
tar_file_serializers	Return registered serializers that operate on gzipped tar archives. TYPE: tuple[Type[IsTarFileSerializer], ...]

Source code in src/omnipy/data/serializer.py

class SerializerRegistry:
    """Registry and auto-detection helper for the serializers available to Omnipy."""
    def __init__(self) -> None:
        self._serializer_classes: list[Type[IsSerializer]] = []

    def register(self, serializer_cls: Type[IsSerializer]) -> None:
        """Register a serializer class for later lookup and auto-detection."""

        self._serializer_classes.append(serializer_cls)

    @property
    def serializers(self) -> tuple[Type[IsSerializer], ...]:
        """Return all registered serializer classes in registration order."""

        return tuple(self._serializer_classes)

    @property
    def tar_file_serializers(self) -> tuple[Type[IsTarFileSerializer], ...]:
        """Return registered serializers that operate on gzipped tar archives."""

        return tuple(cls for cls in self._serializer_classes if issubclass(cls, TarFileSerializer))

    def auto_detect(self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
        """Return the first compatible dataset/serializer pair from the registry."""

        return self._autodetect_serializer(dataset, self.serializers)

    def auto_detect_tar_file_serializer(
            self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
        """Try only tar-file serializers and return the first compatible pair."""

        return self._autodetect_serializer(dataset, self.tar_file_serializers)

    @classmethod
    def _autodetect_serializer(
        cls,
        dataset: IsDataset,
        serializers: tuple[Type[IsSerializer], ...],
    ) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:

        from omnipy.hub.runtime import runtime
        if runtime:
            with hold_and_reset_prev_attrib_value(
                    runtime.config.data.model,
                    'interactive',
            ):
                with hold_and_reset_prev_attrib_value(
                        runtime.config.data.model,
                        'dynamically_convert_elements_to_models',
                ):
                    runtime.config.data.model.interactive = False
                    runtime.config.data.model.dynamically_convert_elements_to_models = False

                    return cls._test_all_serializer_combos(dataset, serializers)
        else:
            return cls._test_all_serializer_combos(dataset, serializers)

    @classmethod
    def _test_all_serializer_combos(
        cls,
        dataset: IsDataset,
        serializers: tuple[Type[IsSerializer], ...],
    ) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:

        # def _direct(dataset: Dataset, serializer: Serializer):
        #     new_dataset_cls = serializer.get_dataset_cls_for_new()
        #     new_dataset = new_dataset_cls(dataset)
        #     return new_dataset

        def _to_data_from_json(dataset: IsDataset, serializer: IsSerializer):
            new_dataset_cls = serializer.get_dataset_cls_for_new()
            new_dataset = new_dataset_cls()
            new_dataset.from_json(dataset.to_data())
            return new_dataset

        def _to_data_from_data(dataset: IsDataset, serializer: IsSerializer):
            new_dataset_cls = serializer.get_dataset_cls_for_new()
            new_dataset = new_dataset_cls()
            new_dataset.from_data(dataset.to_data())
            return new_dataset

        def _to_data_from_data_if_direct(dataset: IsDataset, serializer: IsSerializer):
            assert serializer.is_dataset_directly_supported(dataset)
            return _to_data_from_data(dataset, serializer)

        # def _to_json_from_json(dataset: Dataset, serializer: Serializer):
        #     new_dataset_cls = serializer.get_dataset_cls_for_new()
        #     new_dataset = new_dataset_cls()
        #     new_dataset.from_json(dataset.to_json())
        #     return new_dataset

        for func in (_to_data_from_data_if_direct, _to_data_from_json, _to_data_from_data):
            for serializer in serializers:
                try:
                    new_dataset = func(dataset, serializer)
                    return new_dataset, serializer
                except (TypeError, ValueError, ValidationError, AssertionError):
                    pass

        return None, None

    def detect_tar_file_serializers_from_dataset_cls(
            self, dataset: IsDataset) -> tuple[Type[IsTarFileSerializer], ...]:
        """Return tar-file serializers that can load data into ``dataset``.

        If no direct match exists, serializers using the generic ``bytes`` suffix are returned as a
        fallback.
        """

        serializers = tuple(
            serializer_cls for serializer_cls in self.tar_file_serializers
            if serializer_cls.is_dataset_directly_supported(dataset))
        if len(serializers) == 0:
            serializers = tuple(serializer_cls for serializer_cls in self.tar_file_serializers
                                if serializer_cls.get_output_file_suffix() == 'bytes')
        return serializers

    def detect_tar_file_serializers_from_file_suffix(
            self, file_suffix: str) -> tuple[Type[IsTarFileSerializer], ...]:
        """Return tar-file serializers whose output suffix matches ``file_suffix``."""

        return tuple(serializer_cls for serializer_cls in self.tar_file_serializers
                     if serializer_cls.get_output_file_suffix() == file_suffix)

    def load_from_tar_file_path_based_on_file_suffix(
        self,
        log_obj: CanLog,
        tar_file_path: str,
        to_dataset: IsDataset,
    ) -> IsDataset | None:
        """Load a tar archive by detecting its serializer from member file suffixes.

        Args:
            log_obj: Logger-like object used for status and failure messages.
            tar_file_path: Path to the gzipped tar archive.
            to_dataset: Preferred destination dataset instance.

        Returns:
            The populated destination dataset, an auto-created dataset if conversion fails, or
            ``None`` when no unique serializer can be determined.
        """

        log: Callable
        if hasattr(log_obj, 'log'):
            log = log_obj.log
        else:
            log = print

        with tarfile.open(tar_file_path, 'r:gz') as tarfile_obj:
            file_suffixes = set(fn.split('.')[-1] for fn in tarfile_obj.getnames())
        if len(file_suffixes) != 1:
            log(f'Tar archive contains files with different or '
                f'no file suffixes: {file_suffixes}. Serializer '
                f'cannot be uniquely determined. Aborting '
                f'restore.')
        else:
            file_suffix = file_suffixes.pop()
            serializers = self.detect_tar_file_serializers_from_file_suffix(file_suffix)
            if len(serializers) == 0:
                log(f'No serializer for file suffix "{file_suffix}" can be'
                    f'determined. Aborting restore.')
            else:
                log(f'Reading dataset from a gzipped tarpack at'
                    f' "{os.path.abspath(tar_file_path)}"')

                serializer = serializers[0]
                with open(tar_file_path, 'rb') as tarfile_binary:
                    auto_dataset = serializer.deserialize(tarfile_binary.read())

                if to_dataset.get_type() is auto_dataset.get_type():
                    cast(HasData, to_dataset).data = cast(HasData, auto_dataset).data
                    return to_dataset
                else:
                    try:
                        if to_dataset.get_type().inner_type == str:
                            to_dataset.from_data(auto_dataset.to_json())
                        else:
                            to_dataset.from_json(auto_dataset.to_data())
                        return to_dataset
                    except Exception:
                        return auto_dataset

    def load_from_tar_file_path_based_on_dataset_cls(
        self,
        log_obj: CanLog,
        tar_file_path: str,
        to_dataset: IsDataset,
        any_file_suffix: bool = False,
    ) -> IsDataset | None:
        """Load a tar archive by trying serializers compatible with ``to_dataset``.

        Args:
            log_obj: Logger-like object used for status and failure messages.
            tar_file_path: Path to the gzipped tar archive.
            to_dataset: Dataset instance whose type guides serializer selection.
            any_file_suffix: Whether deserializers may ignore file-suffix checks.

        Returns:
            The first dataset produced by a compatible serializer, or ``None`` if none match.
        """

        log: Callable
        if hasattr(log_obj, 'log'):
            log = log_obj.log
        else:
            log = print

        serializers = self.detect_tar_file_serializers_from_dataset_cls(to_dataset)
        if len(serializers) == 0:
            log(f'No serializer for Dataset with type "{type(to_dataset)}" can be '
                f'determined.')
        else:
            for serializer in serializers:
                log(f'Reading dataset from a gzipped tarpack at'
                    f' "{os.path.abspath(tar_file_path)}" with serializer type: '
                    f'"{serializer.__name__}"')

                with open(tar_file_path, 'rb') as tarfile_binary:
                    out_dataset = serializer.deserialize(
                        tarfile_binary.read(),
                        any_file_suffix=any_file_suffix,
                    )

                return out_dataset

serializers property

serializers: tuple[Type[IsSerializer], ...]

Return all registered serializer classes in registration order.

tar_file_serializers property

tar_file_serializers: tuple[Type[IsTarFileSerializer], ...]

Return registered serializers that operate on gzipped tar archives.

__init__

__init__() -> None

Source code in src/omnipy/data/serializer.py

def __init__(self) -> None:
    self._serializer_classes: list[Type[IsSerializer]] = []

auto_detect

auto_detect(dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]

Return the first compatible dataset/serializer pair from the registry.

Source code in src/omnipy/data/serializer.py

def auto_detect(self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
    """Return the first compatible dataset/serializer pair from the registry."""

    return self._autodetect_serializer(dataset, self.serializers)

auto_detect_tar_file_serializer

auto_detect_tar_file_serializer(
    dataset: IsDataset,
) -> tuple[IsDataset, IsSerializer] | tuple[None, None]

Try only tar-file serializers and return the first compatible pair.

Source code in src/omnipy/data/serializer.py

def auto_detect_tar_file_serializer(
        self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
    """Try only tar-file serializers and return the first compatible pair."""

    return self._autodetect_serializer(dataset, self.tar_file_serializers)

detect_tar_file_serializers_from_dataset_cls

detect_tar_file_serializers_from_dataset_cls(
    dataset: IsDataset,
) -> tuple[Type[IsTarFileSerializer], ...]

Return tar-file serializers that can load data into dataset.

If no direct match exists, serializers using the generic bytes suffix are returned as a fallback.

Source code in src/omnipy/data/serializer.py

def detect_tar_file_serializers_from_dataset_cls(
        self, dataset: IsDataset) -> tuple[Type[IsTarFileSerializer], ...]:
    """Return tar-file serializers that can load data into ``dataset``.

    If no direct match exists, serializers using the generic ``bytes`` suffix are returned as a
    fallback.
    """

    serializers = tuple(
        serializer_cls for serializer_cls in self.tar_file_serializers
        if serializer_cls.is_dataset_directly_supported(dataset))
    if len(serializers) == 0:
        serializers = tuple(serializer_cls for serializer_cls in self.tar_file_serializers
                            if serializer_cls.get_output_file_suffix() == 'bytes')
    return serializers

detect_tar_file_serializers_from_file_suffix

detect_tar_file_serializers_from_file_suffix(
    file_suffix: str,
) -> tuple[Type[IsTarFileSerializer], ...]

Return tar-file serializers whose output suffix matches file_suffix.

Source code in src/omnipy/data/serializer.py

def detect_tar_file_serializers_from_file_suffix(
        self, file_suffix: str) -> tuple[Type[IsTarFileSerializer], ...]:
    """Return tar-file serializers whose output suffix matches ``file_suffix``."""

    return tuple(serializer_cls for serializer_cls in self.tar_file_serializers
                 if serializer_cls.get_output_file_suffix() == file_suffix)

load_from_tar_file_path_based_on_dataset_cls

load_from_tar_file_path_based_on_dataset_cls(
    log_obj: CanLog, tar_file_path: str, to_dataset: IsDataset, any_file_suffix: bool = False
) -> IsDataset | None

Load a tar archive by trying serializers compatible with to_dataset.

PARAMETER	DESCRIPTION
log_obj	Logger-like object used for status and failure messages. TYPE: CanLog
tar_file_path	Path to the gzipped tar archive. TYPE: str
to_dataset	Dataset instance whose type guides serializer selection. TYPE: IsDataset
any_file_suffix	Whether deserializers may ignore file-suffix checks. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
IsDataset | None	The first dataset produced by a compatible serializer, or None if none match.

Source code in src/omnipy/data/serializer.py

def load_from_tar_file_path_based_on_dataset_cls(
    self,
    log_obj: CanLog,
    tar_file_path: str,
    to_dataset: IsDataset,
    any_file_suffix: bool = False,
) -> IsDataset | None:
    """Load a tar archive by trying serializers compatible with ``to_dataset``.

    Args:
        log_obj: Logger-like object used for status and failure messages.
        tar_file_path: Path to the gzipped tar archive.
        to_dataset: Dataset instance whose type guides serializer selection.
        any_file_suffix: Whether deserializers may ignore file-suffix checks.

    Returns:
        The first dataset produced by a compatible serializer, or ``None`` if none match.
    """

    log: Callable
    if hasattr(log_obj, 'log'):
        log = log_obj.log
    else:
        log = print

    serializers = self.detect_tar_file_serializers_from_dataset_cls(to_dataset)
    if len(serializers) == 0:
        log(f'No serializer for Dataset with type "{type(to_dataset)}" can be '
            f'determined.')
    else:
        for serializer in serializers:
            log(f'Reading dataset from a gzipped tarpack at'
                f' "{os.path.abspath(tar_file_path)}" with serializer type: '
                f'"{serializer.__name__}"')

            with open(tar_file_path, 'rb') as tarfile_binary:
                out_dataset = serializer.deserialize(
                    tarfile_binary.read(),
                    any_file_suffix=any_file_suffix,
                )

            return out_dataset

load_from_tar_file_path_based_on_file_suffix

load_from_tar_file_path_based_on_file_suffix(
    log_obj: CanLog, tar_file_path: str, to_dataset: IsDataset
) -> IsDataset | None

Load a tar archive by detecting its serializer from member file suffixes.

PARAMETER	DESCRIPTION
log_obj	Logger-like object used for status and failure messages. TYPE: CanLog
tar_file_path	Path to the gzipped tar archive. TYPE: str
to_dataset	Preferred destination dataset instance. TYPE: IsDataset

RETURNS	DESCRIPTION
IsDataset | None	The populated destination dataset, an auto-created dataset if conversion fails, or None when no unique serializer can be determined.

Source code in src/omnipy/data/serializer.py

def load_from_tar_file_path_based_on_file_suffix(
    self,
    log_obj: CanLog,
    tar_file_path: str,
    to_dataset: IsDataset,
) -> IsDataset | None:
    """Load a tar archive by detecting its serializer from member file suffixes.

    Args:
        log_obj: Logger-like object used for status and failure messages.
        tar_file_path: Path to the gzipped tar archive.
        to_dataset: Preferred destination dataset instance.

    Returns:
        The populated destination dataset, an auto-created dataset if conversion fails, or
        ``None`` when no unique serializer can be determined.
    """

    log: Callable
    if hasattr(log_obj, 'log'):
        log = log_obj.log
    else:
        log = print

    with tarfile.open(tar_file_path, 'r:gz') as tarfile_obj:
        file_suffixes = set(fn.split('.')[-1] for fn in tarfile_obj.getnames())
    if len(file_suffixes) != 1:
        log(f'Tar archive contains files with different or '
            f'no file suffixes: {file_suffixes}. Serializer '
            f'cannot be uniquely determined. Aborting '
            f'restore.')
    else:
        file_suffix = file_suffixes.pop()
        serializers = self.detect_tar_file_serializers_from_file_suffix(file_suffix)
        if len(serializers) == 0:
            log(f'No serializer for file suffix "{file_suffix}" can be'
                f'determined. Aborting restore.')
        else:
            log(f'Reading dataset from a gzipped tarpack at'
                f' "{os.path.abspath(tar_file_path)}"')

            serializer = serializers[0]
            with open(tar_file_path, 'rb') as tarfile_binary:
                auto_dataset = serializer.deserialize(tarfile_binary.read())

            if to_dataset.get_type() is auto_dataset.get_type():
                cast(HasData, to_dataset).data = cast(HasData, auto_dataset).data
                return to_dataset
            else:
                try:
                    if to_dataset.get_type().inner_type == str:
                        to_dataset.from_data(auto_dataset.to_json())
                    else:
                        to_dataset.from_json(auto_dataset.to_data())
                    return to_dataset
                except Exception:
                    return auto_dataset

register

register(serializer_cls: Type[IsSerializer]) -> None

Register a serializer class for later lookup and auto-detection.

Source code in src/omnipy/data/serializer.py

def register(self, serializer_cls: Type[IsSerializer]) -> None:
    """Register a serializer class for later lookup and auto-detection."""

    self._serializer_classes.append(serializer_cls)

TarFileSerializer

Bases: Serializer[_DatasetT], Generic[_DatasetT]

              flowchart BT
              omnipy.data.serializer.TarFileSerializer[TarFileSerializer]
              omnipy.data.serializer.Serializer[Serializer]

                              omnipy.data.serializer.Serializer --> omnipy.data.serializer.TarFileSerializer
                


              click omnipy.data.serializer.TarFileSerializer href "" "omnipy.data.serializer.TarFileSerializer"
              click omnipy.data.serializer.Serializer href "" "omnipy.data.serializer.Serializer"
            

Serializer base class for datasets stored as gzipped tar archives of item files.

Subclassed by:

• Reference Code reference omnipy components
  • json  serializers  JsonDatasetToTarFileSerializer
  • pandas  serializers  PandasDatasetToTarFileSerializer
  • raw  serializers
    •  RawBytesDatasetToTarFileSerializer
    •  RawStrDatasetToTarFileSerializer

METHOD	DESCRIPTION
create_dataset_from_tarfile	Populate dataset from a gzipped tar archive.
create_tarfile_from_dataset	Build a gzipped tar archive by serializing each dataset item into one member file.
deserialize	Deserialize serialized bytes into a dataset instance.
get_dataset_cls_for_new	Return the dataset class created when deserializing with this serializer.
get_output_file_suffix	Return the filename suffix Omnipy associates with this serializer's output.
is_dataset_directly_supported	Return whether dataset can be reconstructed directly by this serializer.
serialize	Serialize dataset into bytes suitable for persistence or transport.

Source code in src/omnipy/data/serializer.py

class TarFileSerializer(Serializer[_DatasetT], Generic[_DatasetT]):
    """Serializer base class for datasets stored as gzipped tar archives of item files."""
    @classmethod
    def create_tarfile_from_dataset(cls,
                                    dataset: _DatasetT,
                                    data_encode_func: Callable[..., bytes | memoryview]) -> bytes:
        """Build a gzipped tar archive by serializing each dataset item into one member file.

        Args:
            dataset: Dataset whose items should be written into the archive.
            data_encode_func: Function converting each dataset item into raw bytes.

        Returns:
            The complete gzipped tar archive as bytes.
        """

        bytes_io = BytesIO()
        with tarfile.open(fileobj=bytes_io, mode='w:gz') as tarfile_stream:
            for data_file, data in dataset.items():  # type: ignore[attr-defined]
                json_data_bytestream = BytesIO(data_encode_func(data))
                json_data_bytestream.seek(0)
                tarinfo = TarInfo(name=f'{data_file}.{cls.get_output_file_suffix()}')
                tarinfo.size = len(json_data_bytestream.getbuffer())
                tarfile_stream.addfile(tarinfo, json_data_bytestream)
        return bytes_io.getbuffer().tobytes()

    @classmethod
    def create_dataset_from_tarfile(cls,
                                    dataset: _DatasetT,
                                    tarfile_bytes: bytes,
                                    data_decode_func: Callable[[IO[bytes]], Any],
                                    dictify_object_func: Callable[[str, Any], dict | str],
                                    import_method: str = 'from_data',
                                    any_file_suffix: bool = False) -> None:
        """Populate ``dataset`` from a gzipped tar archive.

        Args:
            dataset: Dataset instance to populate.
            tarfile_bytes: Serialized gzipped tar archive.
            data_decode_func: Function decoding a single extracted file object.
            dictify_object_func: Function mapping filename and decoded payload to import data.
            import_method: Dataset import method to call for each decoded item.
            any_file_suffix: Whether to skip file-suffix validation inside the archive.
        """

        with tarfile.open(fileobj=BytesIO(tarfile_bytes), mode='r:gz') as tarfile_stream:
            for filename in tarfile_stream.getnames():
                data_file = tarfile_stream.extractfile(filename)
                assert data_file is not None
                if not any_file_suffix:
                    assert filename.endswith(f'.{cls.get_output_file_suffix()}')
                data_file_name = os.path.basename('.'.join(filename.split('.')[:-1]))
                getattr(dataset, import_method)(
                    dictify_object_func(data_file_name, data_decode_func(data_file)))

create_dataset_from_tarfile classmethod

create_dataset_from_tarfile(
    dataset: _DatasetT,
    tarfile_bytes: bytes,
    data_decode_func: Callable[[IO[bytes]], Any],
    dictify_object_func: Callable[[str, Any], dict | str],
    import_method: str = "from_data",
    any_file_suffix: bool = False,
) -> None

Populate dataset from a gzipped tar archive.

PARAMETER	DESCRIPTION
dataset	Dataset instance to populate. TYPE: _DatasetT
tarfile_bytes	Serialized gzipped tar archive. TYPE: bytes
data_decode_func	Function decoding a single extracted file object. TYPE: Callable[[IO[bytes]], Any]
dictify_object_func	Function mapping filename and decoded payload to import data. TYPE: Callable[[str, Any], dict | str]
import_method	Dataset import method to call for each decoded item. TYPE: str DEFAULT: 'from_data'
any_file_suffix	Whether to skip file-suffix validation inside the archive. TYPE: bool DEFAULT: False

Source code in src/omnipy/data/serializer.py

@classmethod
def create_dataset_from_tarfile(cls,
                                dataset: _DatasetT,
                                tarfile_bytes: bytes,
                                data_decode_func: Callable[[IO[bytes]], Any],
                                dictify_object_func: Callable[[str, Any], dict | str],
                                import_method: str = 'from_data',
                                any_file_suffix: bool = False) -> None:
    """Populate ``dataset`` from a gzipped tar archive.

    Args:
        dataset: Dataset instance to populate.
        tarfile_bytes: Serialized gzipped tar archive.
        data_decode_func: Function decoding a single extracted file object.
        dictify_object_func: Function mapping filename and decoded payload to import data.
        import_method: Dataset import method to call for each decoded item.
        any_file_suffix: Whether to skip file-suffix validation inside the archive.
    """

    with tarfile.open(fileobj=BytesIO(tarfile_bytes), mode='r:gz') as tarfile_stream:
        for filename in tarfile_stream.getnames():
            data_file = tarfile_stream.extractfile(filename)
            assert data_file is not None
            if not any_file_suffix:
                assert filename.endswith(f'.{cls.get_output_file_suffix()}')
            data_file_name = os.path.basename('.'.join(filename.split('.')[:-1]))
            getattr(dataset, import_method)(
                dictify_object_func(data_file_name, data_decode_func(data_file)))

create_tarfile_from_dataset classmethod

create_tarfile_from_dataset(
    dataset: _DatasetT, data_encode_func: Callable[..., bytes | memoryview]
) -> bytes

Build a gzipped tar archive by serializing each dataset item into one member file.

PARAMETER	DESCRIPTION
dataset	Dataset whose items should be written into the archive. TYPE: _DatasetT
data_encode_func	Function converting each dataset item into raw bytes. TYPE: Callable[..., bytes | memoryview]

RETURNS	DESCRIPTION
bytes	The complete gzipped tar archive as bytes.

Source code in src/omnipy/data/serializer.py

@classmethod
def create_tarfile_from_dataset(cls,
                                dataset: _DatasetT,
                                data_encode_func: Callable[..., bytes | memoryview]) -> bytes:
    """Build a gzipped tar archive by serializing each dataset item into one member file.

    Args:
        dataset: Dataset whose items should be written into the archive.
        data_encode_func: Function converting each dataset item into raw bytes.

    Returns:
        The complete gzipped tar archive as bytes.
    """

    bytes_io = BytesIO()
    with tarfile.open(fileobj=bytes_io, mode='w:gz') as tarfile_stream:
        for data_file, data in dataset.items():  # type: ignore[attr-defined]
            json_data_bytestream = BytesIO(data_encode_func(data))
            json_data_bytestream.seek(0)
            tarinfo = TarInfo(name=f'{data_file}.{cls.get_output_file_suffix()}')
            tarinfo.size = len(json_data_bytestream.getbuffer())
            tarfile_stream.addfile(tarinfo, json_data_bytestream)
    return bytes_io.getbuffer().tobytes()

deserialize abstractmethod classmethod

deserialize(serialized: bytes, any_file_suffix=False) -> _DatasetT

Deserialize serialized bytes into a dataset instance.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def deserialize(cls, serialized: bytes, any_file_suffix=False) -> _DatasetT:
    """Deserialize ``serialized`` bytes into a dataset instance."""

    pass

get_dataset_cls_for_new abstractmethod classmethod

get_dataset_cls_for_new() -> type[IsDataset]

Return the dataset class created when deserializing with this serializer.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def get_dataset_cls_for_new(cls) -> type[IsDataset]:
    """Return the dataset class created when deserializing with this serializer."""

    pass

get_output_file_suffix abstractmethod classmethod

get_output_file_suffix() -> str

Return the filename suffix Omnipy associates with this serializer's output.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def get_output_file_suffix(cls) -> str:
    """Return the filename suffix Omnipy associates with this serializer's output."""

    pass

is_dataset_directly_supported abstractmethod classmethod

is_dataset_directly_supported(dataset: IsDataset) -> bool

Return whether dataset can be reconstructed directly by this serializer.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    """Return whether ``dataset`` can be reconstructed directly by this serializer."""

    pass

serialize abstractmethod classmethod

serialize(dataset: _DatasetT) -> bytes | memoryview

Serialize dataset into bytes suitable for persistence or transport.

Source code in src/omnipy/data/serializer.py

@classmethod
@abstractmethod
def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
    """Serialize ``dataset`` into bytes suitable for persistence or transport."""

    pass