omnipy.shared.protocols.data

Protocols for Omnipy data models, datasets, serializers, and reactive state.

CLASS	DESCRIPTION
AvailableDisplayDims	Display-space dimensions available for rendering, in pixels.
HasContent	Object with a typed content value that can be read or replaced.
HasData	Object exposing an internal mapping of loaded, pending, and failed items.
IsDataClassCreator	Factory/service protocol that wires config, reactive state, and snapshots.
IsDataset	Dictionary-like collection of named models or nested datasets.
IsFailedData	Metadata describing a dataset entry whose asynchronous load failed.
IsHttpUrlDataset	Dataset protocol for collections of HTTP URL model entries.
IsHttpUrlModel	Model protocol representing a validated HTTP URL value.
IsModel	Single-value data wrapper with typed content and conversion support.
IsMultiModelDataset	Dataset protocol that can assign different model classes per item.
IsPendingData	Metadata describing a dataset entry that is still processing.
IsReactive	Mutable reactive value wrapper used for config/state propagation.
IsReactiveObjects	Bundle of reactive objects shared by display and configuration layers.
IsSerializer	Serializer interface for converting datasets to and from bytes.
IsSerializerRegistry	Registry that tracks serializers and selects suitable ones for datasets.
IsSnapshotHolder	Container protocol managing snapshots used for deepcopy/reactive tracking.
IsSnapshotWrapper	Snapshot record linking an object identity to captured content.
IsTarFileSerializer	Serializer extension that stores dataset entries inside tar archives.

ATTRIBUTE	DESCRIPTION
ContentT
HasContentT
IsPathOrUrl	TYPE: TypeAlias
IsPathsOrUrls	TYPE: TypeAlias
IsPathsOrUrlsOneOrMore	TYPE: TypeAlias
IsPathsOrUrlsOneOrMoreOrNone	TYPE: TypeAlias
ObjContraT

ContentT module-attribute

ContentT = TypeVar('ContentT', bound=object)

Returned by:

• Reference Code reference omnipy
  • data
    •  dataset  Dataset  snapshot_holder
    •  model  Model  snapshot_holder
    •  multi  MultiModelDataset  snapshot_holder
    •  snapshot  SnapshotWrapper  snapshot
  • shared protocols  data
    •  HasContent  content
    •  IsDataClassCreator  snapshot_holder
    •  IsHttpUrlModel  content
    •  IsModel  content
    •  IsReactive  value
    •  IsSnapshotWrapper  snapshot

Subclassed by:

• Reference Code reference omnipy
  • data  snapshot
    •  SnapshotHolder
    •  SnapshotWrapper
  • shared protocols  data
    •  HasContent
    •  IsDataClassCreator
    •  IsReactive
    •  IsSnapshotHolder
    •  IsSnapshotWrapper

Used by:

• Reference Code reference omnipy
  • data  snapshot  SnapshotWrapper  __init__
  • shared protocols  data  IsReactive  set

HasContentT module-attribute

HasContentT = TypeVar('HasContentT', bound='HasContent')

Returned by:

• Reference Code reference omnipy shared protocols  data  IsDataClassCreator  snapshot_holder

Subclassed by:

• Reference Code reference omnipy
  • data  snapshot  SnapshotHolder
  • shared protocols  data
    •  IsDataClassCreator
    •  IsSnapshotHolder

Used by:

• Reference Code reference omnipy
  • data  snapshot  SnapshotHolder  take_snapshot
  • shared protocols  data  IsSnapshotHolder  take_snapshot

IsPathOrUrl module-attribute

IsPathOrUrl: TypeAlias = 'str | IsHttpUrlModel'

Used by:

• Reference Code reference
  •  omnipy  Dataset
    •  load
    •  load_into
  • omnipy
    • data
      •  dataset  Dataset
        •  load
        •  load_into
      •  multi  MultiModelDataset
        •  load
        •  load_into
    • shared protocols  data
      •  IsDataset
        •  load
        •  load_into
      •  IsHttpUrlDataset
        •  load
        •  load_into
      •  IsMultiModelDataset
        •  load
        •  load_into

IsPathsOrUrls module-attribute

IsPathsOrUrls: TypeAlias = 'Iterable[str] | IsHttpUrlDataset | Mapping[str, IsPathOrUrl]'

IsPathsOrUrlsOneOrMore module-attribute

IsPathsOrUrlsOneOrMore: TypeAlias = 'IsPathOrUrl | IsPathsOrUrls'

IsPathsOrUrlsOneOrMoreOrNone module-attribute

IsPathsOrUrlsOneOrMoreOrNone: TypeAlias = 'IsPathsOrUrlsOneOrMore | None'

Used by:

• Reference Code reference
  •  omnipy  Dataset
    •  load
    •  load_into
  • omnipy
    • data
      •  dataset  Dataset
        •  load
        •  load_into
      •  multi  MultiModelDataset
        •  load
        •  load_into
    • shared protocols  data
      •  IsDataset
        •  load
        •  load_into
      •  IsHttpUrlDataset
        •  load
        •  load_into
      •  IsMultiModelDataset
        •  load
        •  load_into

ObjContraT module-attribute

ObjContraT = TypeVar('ObjContraT', contravariant=True, bound=object)

Subclassed by:

• Reference Code reference omnipy
  • data  snapshot  SnapshotWrapper
  • shared protocols  data  IsSnapshotWrapper

Used by:

• Reference Code reference omnipy
  • data  snapshot  SnapshotWrapper
    •  differs_from
    •  taken_of_same_obj
  • shared protocols  data  IsSnapshotWrapper
    •  differs_from
    •  taken_of_same_obj

AvailableDisplayDims

Bases: TypedDict

              flowchart BT
              omnipy.shared.protocols.data.AvailableDisplayDims[AvailableDisplayDims]

              

              click omnipy.shared.protocols.data.AvailableDisplayDims href "" "omnipy.shared.protocols.data.AvailableDisplayDims"
            

Display-space dimensions available for rendering, in pixels.

Returned by:

• Reference Code reference omnipy shared protocols  data  IsReactiveObjects  available_display_dims_in_px

ATTRIBUTE	DESCRIPTION
height	TYPE: pyd.NonNegativeInt | None
width	TYPE: pyd.NonNegativeInt | None

Source code in src/omnipy/shared/protocols/data.py

class AvailableDisplayDims(TypedDict):
    """Display-space dimensions available for rendering, in pixels."""

    width: pyd.NonNegativeInt | None
    height: pyd.NonNegativeInt | None

height instance-attribute

height: pyd.NonNegativeInt | None

width instance-attribute

width: pyd.NonNegativeInt | None

HasContent

Bases: Protocol[ContentT]

              flowchart BT
              omnipy.shared.protocols.data.HasContent[HasContent]

              

              click omnipy.shared.protocols.data.HasContent href "" "omnipy.shared.protocols.data.HasContent"
            

Object with a typed content value that can be read or replaced.

Returned by:

• Reference Code reference omnipy data
  •  dataset  Dataset  snapshot_holder
  •  model  Model  snapshot_holder
  •  multi  MultiModelDataset  snapshot_holder

Subclassed by:

• Reference Code reference omnipy shared protocols  data  IsModel

ATTRIBUTE	DESCRIPTION
content	Return the wrapped content value. TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class HasContent(Protocol[ContentT]):
    """Object with a typed ``content`` value that can be read or replaced."""
    @property
    def content(self) -> ContentT:
        """Return the wrapped content value.

        Returns:
            ContentT: Current content stored by the object.
        """
        ...

    @content.setter
    def content(self, value: ContentT) -> None:
        """Replace the wrapped content value.

        Args:
            value: New content to store.
        """
        ...

content property writable

content: ContentT

Return the wrapped content value.

RETURNS	DESCRIPTION
ContentT	Current content stored by the object. TYPE: ContentT

HasData

Bases: Protocol

              flowchart BT
              omnipy.shared.protocols.data.HasData[HasData]

              

              click omnipy.shared.protocols.data.HasData href "" "omnipy.shared.protocols.data.HasData"
            

Object exposing an internal mapping of loaded, pending, and failed items.

ATTRIBUTE	DESCRIPTION
data	TYPE: dict[str, Any | IsPendingData | IsFailedData]

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class HasData(Protocol):
    """Object exposing an internal mapping of loaded, pending, and failed items."""

    data: dict[str, Any | IsPendingData | IsFailedData]

data instance-attribute

data: dict[str, Any | IsPendingData | IsFailedData]

IsDataClassCreator

Bases: Protocol[HasContentT, ContentT]

              flowchart BT
              omnipy.shared.protocols.data.IsDataClassCreator[IsDataClassCreator]

              

              click omnipy.shared.protocols.data.IsDataClassCreator href "" "omnipy.shared.protocols.data.IsDataClassCreator"
            

Factory/service protocol that wires config, reactive state, and snapshots.

Returned by:

• Reference Code reference omnipy
  • data  model  ModelMetaclass  data_class_creator
  • hub  runtime  RuntimeObjects  data_class_creator
  • shared protocols hub  runtime  IsRuntimeObjects  data_class_creator

METHOD	DESCRIPTION
deepcopy_context	Return a context manager for coordinated deepcopy bookkeeping.
set_config	Replace the shared data configuration for the owning data-class family.
set_reactive_objects	Store the shared bundle of reactive runtime objects.

ATTRIBUTE	DESCRIPTION
config	Return the data configuration shared by the owning data-class family. TYPE: IsDataConfig
reactive_objects	Return the shared bundle of reactive runtime objects, if configured. TYPE: IsReactiveObjects | None
snapshot_holder	Return the snapshot manager used for deepcopy and reactive tracking. TYPE: IsSnapshotHolder[HasContentT, ContentT]

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsDataClassCreator(Protocol[HasContentT, ContentT]):
    """Factory/service protocol that wires config, reactive state, and snapshots."""
    @property
    def config(self) -> IsDataConfig:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISDATACLASSCREATOR_CONFIG_SUMMARY}}
        #
        # {{ISDATACLASSCREATOR_CONFIG_DETAILS}}
        """Return the data configuration shared by the owning data-class family.

        Returns:
            IsDataConfig: Shared configuration object used by related models and datasets.
        """
        ...

    def set_config(self, config: IsDataConfig) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISDATACLASSCREATOR_SET_CONFIG_SUMMARY}}
        #
        # {{ISDATACLASSCREATOR_SET_CONFIG_DETAILS}}
        """Replace the shared data configuration for the owning data-class family.

        Args:
            config: Data configuration object to store for related models and datasets.
        """
        ...

    @property
    def reactive_objects(self) -> IsReactiveObjects | None:
        """Return the shared bundle of reactive runtime objects, if configured.

        Returns:
            IsReactiveObjects | None: Shared reactive-object bundle, or ``None``.
        """
        ...

    def set_reactive_objects(self, reactive_objects: IsReactiveObjects) -> None:
        """Store the shared bundle of reactive runtime objects.

        Args:
            reactive_objects: Reactive-object bundle to share with related data classes.
        """
        ...

    @property
    def snapshot_holder(self) -> IsSnapshotHolder[HasContentT, ContentT]:
        """Return the snapshot manager used for deepcopy and reactive tracking.

        Returns:
            IsSnapshotHolder[HasContentT, ContentT]: Snapshot manager for related objects.
        """
        ...

    def deepcopy_context(
        self,
        top_level_entry_func: Callable[[], None],
        top_level_exit_func: Callable[[], None],
    ) -> ContextManager[int]:
        """Return a context manager for coordinated deepcopy bookkeeping.

        Args:
            top_level_entry_func: Callback invoked when entering the outermost deepcopy.
            top_level_exit_func: Callback invoked when leaving the outermost deepcopy.

        Returns:
            ContextManager[int]: Context manager tracking nested deepcopy depth.
        """
        ...

config property

config: IsDataConfig

Return the data configuration shared by the owning data-class family.

RETURNS	DESCRIPTION
IsDataConfig	Shared configuration object used by related models and datasets. TYPE: IsDataConfig

reactive_objects property

reactive_objects: IsReactiveObjects | None

Return the shared bundle of reactive runtime objects, if configured.

RETURNS	DESCRIPTION
IsReactiveObjects | None	IsReactiveObjects | None: Shared reactive-object bundle, or None.

snapshot_holder property

snapshot_holder: IsSnapshotHolder[HasContentT, ContentT]

Return the snapshot manager used for deepcopy and reactive tracking.

RETURNS	DESCRIPTION
IsSnapshotHolder[HasContentT, ContentT]	IsSnapshotHolder[HasContentT, ContentT]: Snapshot manager for related objects.

deepcopy_context

deepcopy_context(
    top_level_entry_func: Callable[[], None], top_level_exit_func: Callable[[], None]
) -> ContextManager[int]

Return a context manager for coordinated deepcopy bookkeeping.

PARAMETER	DESCRIPTION
top_level_entry_func	Callback invoked when entering the outermost deepcopy. TYPE: Callable[[], None]
top_level_exit_func	Callback invoked when leaving the outermost deepcopy. TYPE: Callable[[], None]

RETURNS	DESCRIPTION
ContextManager[int]	ContextManager[int]: Context manager tracking nested deepcopy depth.

Source code in src/omnipy/shared/protocols/data.py

def deepcopy_context(
    self,
    top_level_entry_func: Callable[[], None],
    top_level_exit_func: Callable[[], None],
) -> ContextManager[int]:
    """Return a context manager for coordinated deepcopy bookkeeping.

    Args:
        top_level_entry_func: Callback invoked when entering the outermost deepcopy.
        top_level_exit_func: Callback invoked when leaving the outermost deepcopy.

    Returns:
        ContextManager[int]: Context manager tracking nested deepcopy depth.
    """
    ...

set_config

set_config(config: IsDataConfig) -> None

Replace the shared data configuration for the owning data-class family.

PARAMETER	DESCRIPTION
config	Data configuration object to store for related models and datasets. TYPE: IsDataConfig

Source code in src/omnipy/shared/protocols/data.py

def set_config(self, config: IsDataConfig) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISDATACLASSCREATOR_SET_CONFIG_SUMMARY}}
    #
    # {{ISDATACLASSCREATOR_SET_CONFIG_DETAILS}}
    """Replace the shared data configuration for the owning data-class family.

    Args:
        config: Data configuration object to store for related models and datasets.
    """
    ...

set_reactive_objects

set_reactive_objects(reactive_objects: IsReactiveObjects) -> None

Store the shared bundle of reactive runtime objects.

PARAMETER	DESCRIPTION
reactive_objects	Reactive-object bundle to share with related data classes. TYPE: IsReactiveObjects

Source code in src/omnipy/shared/protocols/data.py

def set_reactive_objects(self, reactive_objects: IsReactiveObjects) -> None:
    """Store the shared bundle of reactive runtime objects.

    Args:
        reactive_objects: Reactive-object bundle to share with related data classes.
    """
    ...

IsDataset

Bases: IsMutableMapping[str, _ModelOrDatasetT], Protocol[_ModelOrDatasetT]

              flowchart BT
              omnipy.shared.protocols.data.IsDataset[IsDataset]
              omnipy.shared.protocols.typing.IsMutableMapping[IsMutableMapping]
              omnipy.shared.protocols.typing.IsMapping[IsMapping]

                              omnipy.shared.protocols.typing.IsMutableMapping --> omnipy.shared.protocols.data.IsDataset
                                omnipy.shared.protocols.typing.IsMapping --> omnipy.shared.protocols.typing.IsMutableMapping
                



              click omnipy.shared.protocols.data.IsDataset href "" "omnipy.shared.protocols.data.IsDataset"
              click omnipy.shared.protocols.typing.IsMutableMapping href "" "omnipy.shared.protocols.typing.IsMutableMapping"
              click omnipy.shared.protocols.typing.IsMapping href "" "omnipy.shared.protocols.typing.IsMapping"
            

Dictionary-like collection of named models or nested datasets.

Datasets expose conversion helpers for plain data, JSON, and file-based load/save operations.

Returned by:

• Reference Code reference omnipy
  • components
    • json  serializers  JsonDatasetToTarFileSerializer  get_dataset_cls_for_new
    • pandas  serializers  PandasDatasetToTarFileSerializer  get_dataset_cls_for_new
    • raw  serializers
      •  RawBytesDatasetToTarFileSerializer  get_dataset_cls_for_new
      •  RawStrDatasetToTarFileSerializer  get_dataset_cls_for_new
  • data  serializer
    •  Serializer  get_dataset_cls_for_new
    •  SerializerRegistry
      •  auto_detect
      •  auto_detect_tar_file_serializer
      •  load_from_tar_file_path_based_on_dataset_cls
      •  load_from_tar_file_path_based_on_file_suffix
    •  TarFileSerializer  get_dataset_cls_for_new
  • shared protocols
    •  data
      •  IsSerializer  get_dataset_cls_for_new
      •  IsSerializerRegistry
        •  auto_detect
        •  auto_detect_tar_file_serializer
        •  load_from_tar_file_path_based_on_dataset_cls
        •  load_from_tar_file_path_based_on_file_suffix
      •  IsTarFileSerializer  get_dataset_cls_for_new
    • compute  job
      •  IsAnyFlow  output_dataset_cls
      •  IsAnyFlowTemplate  output_dataset_cls
      •  IsChildJobListArgJob  output_dataset_cls
      •  IsChildJobListArgJobBase  output_dataset_cls
      •  IsChildJobListArgJobTemplate  output_dataset_cls
      •  IsDagFlow  output_dataset_cls
      •  IsDagFlowTemplate  output_dataset_cls
      •  IsFuncArgJob  output_dataset_cls
      •  IsFuncArgJobBase  output_dataset_cls
      •  IsFuncArgJobTemplate  output_dataset_cls
      •  IsFuncFlow  output_dataset_cls
      •  IsFuncFlowTemplate  output_dataset_cls
      •  IsLinearFlow  output_dataset_cls
      •  IsLinearFlowTemplate  output_dataset_cls
      •  IsTask  output_dataset_cls
      •  IsTaskTemplate  output_dataset_cls

Subclassed by:

• Reference Code reference omnipy shared protocols  data
  •  IsHttpUrlDataset
  •  IsMultiModelDataset

Used by:

• Reference Code reference omnipy
  • components
    • json  serializers  JsonDatasetToTarFileSerializer  is_dataset_directly_supported
    • pandas  serializers  PandasDatasetToTarFileSerializer  is_dataset_directly_supported
    • raw  serializers
      •  RawBytesDatasetToTarFileSerializer  is_dataset_directly_supported
      •  RawStrDatasetToTarFileSerializer  is_dataset_directly_supported
  • data  serializer
    •  Serializer  is_dataset_directly_supported
    •  SerializerRegistry
      •  auto_detect
      •  auto_detect_tar_file_serializer
      •  detect_tar_file_serializers_from_dataset_cls
      •  load_from_tar_file_path_based_on_dataset_cls
      •  load_from_tar_file_path_based_on_file_suffix
    •  TarFileSerializer  is_dataset_directly_supported
  • shared protocols
    •  data
      •  IsSerializer  is_dataset_directly_supported
      •  IsSerializerRegistry
        •  auto_detect
        •  auto_detect_tar_file_serializer
        •  detect_tar_file_serializers_from_dataset_cls
        •  load_from_tar_file_path_based_on_dataset_cls
        •  load_from_tar_file_path_based_on_file_suffix
      •  IsTarFileSerializer  is_dataset_directly_supported
    • compute  job
      •  IsAnyFlowTemplate  refine
      •  IsChildJobListArgJobTemplate  refine
      •  IsDagFlowTemplate  refine
      •  IsFuncArgJobTemplate  refine
      •  IsFuncFlowTemplate  refine
      •  IsLinearFlowTemplate  refine
      •  IsTaskTemplate  refine

METHOD	DESCRIPTION
__init__
clear	D.clear() -> None. Remove all items from D.
failed_task_details	Return failure metadata for entries whose processing failed.
from_data	Populate the dataset from plain Python data.
from_json	Populate the dataset from JSON-encoded entries.
get	D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.
get_type	Return the item type stored by this dataset class.
keys	D.keys() -> a set-like object providing a view on D's keys
load	Load dataset content from one or more paths or URLs.
load_into	Load external content into the current dataset instance.
pending_task_details	Return metadata for entries that are still processing.
pop	D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
popitem	D.popitem() -> (k, v), remove and return some (key, value) pair
save	Persist the dataset to a filesystem path.
setdefault	D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
to_data	Convert the dataset to plain Python data.
to_json	Serialize the dataset to JSON strings.
to_json_schema	Return the JSON schema for this dataset type.
update	D.update([E, ]**F) -> None. Update D from mapping/iterable E and F.
values	D.values() -> an object providing a view on D's values

ATTRIBUTE	DESCRIPTION
available_data	Return a view containing only successfully available entries. TYPE: Self
failed_data	Return a view containing entries whose processing failed. TYPE: Self
pending_data	Return a view containing entries that are still processing. TYPE: Self

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsDataset(IsMutableMapping[str, _ModelOrDatasetT], Protocol[_ModelOrDatasetT]):
    """Dictionary-like collection of named models or nested datasets.

    Datasets expose conversion helpers for plain data, JSON, and file-based
    load/save operations.
    """
    def __init__(
        self,
        value: Mapping[str, object] | Iterator[tuple[str, object]] | UndefinedType = Undefined,
        *,
        data: Mapping[str, object] | UndefinedType = Undefined,
        **input_data: object,
    ) -> None:
        ...

    @classmethod
    @functools.cache
    def get_type(cls) -> type[_ModelOrDatasetT]:
        """Return the item type stored by this dataset class.

        Returns:
            type[_ModelOrDatasetT]: Model or nested-dataset type used for items.
        """
        ...

    def to_data(self) -> dict[str, Any]:
        """Convert the dataset to plain Python data.

        Returns:
            dict[str, Any]: Plain-data representation keyed by dataset entry name.
        """
        ...

    def from_data(self,
                  data: Mapping[str, Any] | Iterable[tuple[str, Any]],
                  update: bool = True) -> None:
        """Populate the dataset from plain Python data.

        Args:
            data: Plain-data mapping or iterable of key-value pairs to import.
            update: Whether imported values should merge into existing content.
        """
        ...

    def to_json(self, pretty=True) -> dict[str, str]:
        """Serialize the dataset to JSON strings.

        Args:
            pretty: Whether to format the JSON output for readability.

        Returns:
            dict[str, str]: JSON representation for each dataset entry.
        """
        ...

    def from_json(self,
                  data: Mapping[str, str] | Iterable[tuple[str, str]],
                  update: bool = True) -> None:
        """Populate the dataset from JSON-encoded entries.

        Args:
            data: JSON strings keyed by dataset entry name.
            update: Whether imported values should merge into existing content.
        """
        ...

    @classmethod
    def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
        """Return the JSON schema for this dataset type.

        Args:
            pretty: Whether to format the schema output for readability.

        Returns:
            str | dict[str, str]: JSON schema as one string or per-entry mapping.
        """
        ...

    def save(self, path: str) -> None:
        """Persist the dataset to a filesystem path.

        Args:
            path: Destination path for the serialized dataset.
        """
        ...

    @classmethod
    def load(
        cls,
        paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
        by_file_suffix: bool = False,
        as_mime_type: None | str = None,
        **kwargs: IsPathOrUrl,
    ) -> Self | asyncio.Task[Self]:
        """Load dataset content from one or more paths or URLs.

        Args:
            paths_or_urls: Source path, URL, or collection of sources to load.
            by_file_suffix: Whether serializer lookup should prefer file suffixes.
            as_mime_type: Explicit MIME type override, if any.
            kwargs: Additional named path or URL sources.

        Returns:
            Self | asyncio.Task[Self]: Loaded dataset or asynchronous load task.
        """
        ...

    def load_into(
        self,
        paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
        by_file_suffix: bool = False,
        as_mime_type: None | str = None,
        **kwargs: IsPathOrUrl,
    ) -> Self | asyncio.Task[Self]:
        """Load external content into the current dataset instance.

        Args:
            paths_or_urls: Source path, URL, or collection of sources to load.
            by_file_suffix: Whether serializer lookup should prefer file suffixes.
            as_mime_type: Explicit MIME type override, if any.
            kwargs: Additional named path or URL sources.

        Returns:
            Self | asyncio.Task[Self]: Updated dataset or asynchronous load task.
        """
        ...

    @property
    def available_data(self) -> Self:
        """Return a view containing only successfully available entries.

        Returns:
            Self: Dataset containing entries whose content is already available.
        """
        ...

    @property
    def pending_data(self) -> Self:
        """Return a view containing entries that are still processing.

        Returns:
            Self: Dataset containing entries backed by pending work.
        """
        ...

    @property
    def failed_data(self) -> Self:
        """Return a view containing entries whose processing failed.

        Returns:
            Self: Dataset containing entries associated with failures.
        """
        ...

    def pending_task_details(self) -> dict[str, IsPendingData]:
        """Return metadata for entries that are still processing.

        Returns:
            dict[str, IsPendingData]: Pending metadata keyed by dataset entry name.
        """
        ...

    def failed_task_details(self) -> dict[str, IsFailedData]:
        """Return failure metadata for entries whose processing failed.

        Returns:
            dict[str, IsFailedData]: Failure metadata keyed by dataset entry name.
        """
        ...

    # TODO: Remove methods of IsDataset that overlap with IsMutableMapping?

    @overload
    def __getitem__(self, selector: str | int) -> _ModelOrDatasetT:
        ...

    @overload
    def __getitem__(self, selector: slice | Iterable[str | int]) -> Self:
        ...

    @override
    def __getitem__(self,
                    selector: str | int | slice | Iterable[str | int]) -> '_ModelOrDatasetT | Self':
        ...

    @overload
    def __setitem__(self, selector: str | int, data_obj: _ModelOrDatasetT) -> None:
        ...

    @overload
    def __setitem__(self,
                    selector: slice | Iterable[str | int],
                    data_obj: Mapping[str, _ModelOrDatasetT] | Iterable[_ModelOrDatasetT]) -> None:
        ...

    def __setitem__(
        self,
        selector: str | int | slice | Iterable[str | int],
        data_obj: _ModelOrDatasetT | Mapping[str, _ModelOrDatasetT] | Iterable[_ModelOrDatasetT],
    ) -> None:
        ...

    def __delitem__(self, selector: str | int | slice | Iterable[str | int]) -> None:
        ...

available_data property

available_data: Self

Return a view containing only successfully available entries.

RETURNS	DESCRIPTION
Self	Dataset containing entries whose content is already available. TYPE: Self

failed_data property

failed_data: Self

Return a view containing entries whose processing failed.

RETURNS	DESCRIPTION
Self	Dataset containing entries associated with failures. TYPE: Self

pending_data property

pending_data: Self

Return a view containing entries that are still processing.

RETURNS	DESCRIPTION
Self	Dataset containing entries backed by pending work. TYPE: Self

__init__

__init__(
    value: Mapping[str, object] | Iterator[tuple[str, object]] | UndefinedType = Undefined,
    *,
    data: Mapping[str, object] | UndefinedType = Undefined,
    **input_data: object,
) -> None

Source code in src/omnipy/shared/protocols/data.py

def __init__(
    self,
    value: Mapping[str, object] | Iterator[tuple[str, object]] | UndefinedType = Undefined,
    *,
    data: Mapping[str, object] | UndefinedType = Undefined,
    **input_data: object,
) -> None:
    ...

clear

clear() -> None

D.clear() -> None. Remove all items from D.

Source code in src/omnipy/shared/protocols/typing.py

def clear(self) -> None:
    """
    D.clear() -> None.  Remove all items from D.
    """
    raise AssumedToBeImplementedException

failed_task_details

failed_task_details() -> dict[str, IsFailedData]

Return failure metadata for entries whose processing failed.

RETURNS	DESCRIPTION
dict[str, IsFailedData]	dict[str, IsFailedData]: Failure metadata keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def failed_task_details(self) -> dict[str, IsFailedData]:
    """Return failure metadata for entries whose processing failed.

    Returns:
        dict[str, IsFailedData]: Failure metadata keyed by dataset entry name.
    """
    ...

from_data

from_data(data: Mapping[str, Any] | Iterable[tuple[str, Any]], update: bool = True) -> None

Populate the dataset from plain Python data.

PARAMETER	DESCRIPTION
data	Plain-data mapping or iterable of key-value pairs to import. TYPE: Mapping[str, Any] | Iterable[tuple[str, Any]]
update	Whether imported values should merge into existing content. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/data.py

def from_data(self,
              data: Mapping[str, Any] | Iterable[tuple[str, Any]],
              update: bool = True) -> None:
    """Populate the dataset from plain Python data.

    Args:
        data: Plain-data mapping or iterable of key-value pairs to import.
        update: Whether imported values should merge into existing content.
    """
    ...

from_json

from_json(data: Mapping[str, str] | Iterable[tuple[str, str]], update: bool = True) -> None

Populate the dataset from JSON-encoded entries.

PARAMETER	DESCRIPTION
data	JSON strings keyed by dataset entry name. TYPE: Mapping[str, str] | Iterable[tuple[str, str]]
update	Whether imported values should merge into existing content. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/data.py

def from_json(self,
              data: Mapping[str, str] | Iterable[tuple[str, str]],
              update: bool = True) -> None:
    """Populate the dataset from JSON-encoded entries.

    Args:
        data: JSON strings keyed by dataset entry name.
        update: Whether imported values should merge into existing content.
    """
    ...

get

get(key: _KT) -> _VT_co | None

get(key: _KT, default: _VT_co) -> _VT_co

get(key: _KT, default: _T) -> _VT_co | _T

D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.

Source code in src/omnipy/shared/protocols/typing.py

def get(self, key: _KT, default: None | _VT_co | _T = None, /) -> _VT_co | _T | None:
    """
    D.get(k[,d]) -> D[k] if k in D, else d.  d defaults to None.
    """
    raise AssumedToBeImplementedException

get_type cached classmethod

get_type() -> type[_ModelOrDatasetT]

Return the item type stored by this dataset class.

RETURNS	DESCRIPTION
type[_ModelOrDatasetT]	type[_ModelOrDatasetT]: Model or nested-dataset type used for items.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
@functools.cache
def get_type(cls) -> type[_ModelOrDatasetT]:
    """Return the item type stored by this dataset class.

    Returns:
        type[_ModelOrDatasetT]: Model or nested-dataset type used for items.
    """
    ...

keys

keys() -> IsKeysView[_KT]

D.keys() -> a set-like object providing a view on D's keys

Source code in src/omnipy/shared/protocols/typing.py

def keys(self) -> IsKeysView[_KT]:
    """
    D.keys() -> a set-like object providing a view on D's keys
    """
    raise AssumedToBeImplementedException

load classmethod

load(
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]

Load dataset content from one or more paths or URLs.

PARAMETER	DESCRIPTION
paths_or_urls	Source path, URL, or collection of sources to load. TYPE: IsPathsOrUrlsOneOrMoreOrNone DEFAULT: None
by_file_suffix	Whether serializer lookup should prefer file suffixes. TYPE: bool DEFAULT: False
as_mime_type	Explicit MIME type override, if any. TYPE: None | str DEFAULT: None
kwargs	Additional named path or URL sources. TYPE: IsPathOrUrl DEFAULT: {}

RETURNS	DESCRIPTION
Self | asyncio.Task[Self]	Self | asyncio.Task[Self]: Loaded dataset or asynchronous load task.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def load(
    cls,
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]:
    """Load dataset content from one or more paths or URLs.

    Args:
        paths_or_urls: Source path, URL, or collection of sources to load.
        by_file_suffix: Whether serializer lookup should prefer file suffixes.
        as_mime_type: Explicit MIME type override, if any.
        kwargs: Additional named path or URL sources.

    Returns:
        Self | asyncio.Task[Self]: Loaded dataset or asynchronous load task.
    """
    ...

load_into

load_into(
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]

Load external content into the current dataset instance.

PARAMETER	DESCRIPTION
paths_or_urls	Source path, URL, or collection of sources to load. TYPE: IsPathsOrUrlsOneOrMoreOrNone DEFAULT: None
by_file_suffix	Whether serializer lookup should prefer file suffixes. TYPE: bool DEFAULT: False
as_mime_type	Explicit MIME type override, if any. TYPE: None | str DEFAULT: None
kwargs	Additional named path or URL sources. TYPE: IsPathOrUrl DEFAULT: {}

RETURNS	DESCRIPTION
Self | asyncio.Task[Self]	Self | asyncio.Task[Self]: Updated dataset or asynchronous load task.

Source code in src/omnipy/shared/protocols/data.py

def load_into(
    self,
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]:
    """Load external content into the current dataset instance.

    Args:
        paths_or_urls: Source path, URL, or collection of sources to load.
        by_file_suffix: Whether serializer lookup should prefer file suffixes.
        as_mime_type: Explicit MIME type override, if any.
        kwargs: Additional named path or URL sources.

    Returns:
        Self | asyncio.Task[Self]: Updated dataset or asynchronous load task.
    """
    ...

pending_task_details

pending_task_details() -> dict[str, IsPendingData]

Return metadata for entries that are still processing.

RETURNS	DESCRIPTION
dict[str, IsPendingData]	dict[str, IsPendingData]: Pending metadata keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def pending_task_details(self) -> dict[str, IsPendingData]:
    """Return metadata for entries that are still processing.

    Returns:
        dict[str, IsPendingData]: Pending metadata keyed by dataset entry name.
    """
    ...

pop

pop(key: _KT) -> _VT

pop(key: _KT, default: _VT) -> _VT

pop(key: _KT, default: _T) -> _VT | _T

D.pop(k[,d]) -> v, remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised.

Source code in src/omnipy/shared/protocols/typing.py

def pop(self, key: _KT, default: None | _VT | _T = None, /) -> _VT | _T | None:
    """
    D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
    If key is not found, d is returned if given, otherwise KeyError is raised.
    """
    raise AssumedToBeImplementedException

popitem

popitem() -> tuple[_KT, _VT]

D.popitem() -> (k, v), remove and return some (key, value) pair as a 2-tuple; but raise KeyError if D is empty.

Source code in src/omnipy/shared/protocols/typing.py

def popitem(self) -> tuple[_KT, _VT]:
    """
    D.popitem() -> (k, v), remove and return some (key, value) pair
       as a 2-tuple; but raise KeyError if D is empty.
    """
    raise AssumedToBeImplementedException

save

save(path: str) -> None

Persist the dataset to a filesystem path.

PARAMETER	DESCRIPTION
path	Destination path for the serialized dataset. TYPE: str

Source code in src/omnipy/shared/protocols/data.py

def save(self, path: str) -> None:
    """Persist the dataset to a filesystem path.

    Args:
        path: Destination path for the serialized dataset.
    """
    ...

setdefault

setdefault(key: _KT, default: _VT) -> _VT

D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D

Source code in src/omnipy/shared/protocols/typing.py

def setdefault(self, key: _KT, default: _VT, /) -> _VT:
    """
    D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
    """
    raise AssumedToBeImplementedException

to_data

to_data() -> dict[str, Any]

Convert the dataset to plain Python data.

RETURNS	DESCRIPTION
dict[str, Any]	dict[str, Any]: Plain-data representation keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def to_data(self) -> dict[str, Any]:
    """Convert the dataset to plain Python data.

    Returns:
        dict[str, Any]: Plain-data representation keyed by dataset entry name.
    """
    ...

to_json

to_json(pretty=True) -> dict[str, str]

Serialize the dataset to JSON strings.

PARAMETER	DESCRIPTION
pretty	Whether to format the JSON output for readability. DEFAULT: True

RETURNS	DESCRIPTION
dict[str, str]	dict[str, str]: JSON representation for each dataset entry.

Source code in src/omnipy/shared/protocols/data.py

def to_json(self, pretty=True) -> dict[str, str]:
    """Serialize the dataset to JSON strings.

    Args:
        pretty: Whether to format the JSON output for readability.

    Returns:
        dict[str, str]: JSON representation for each dataset entry.
    """
    ...

to_json_schema classmethod

to_json_schema(pretty=True) -> str | dict[str, str]

Return the JSON schema for this dataset type.

PARAMETER	DESCRIPTION
pretty	Whether to format the schema output for readability. DEFAULT: True

RETURNS	DESCRIPTION
str | dict[str, str]	str | dict[str, str]: JSON schema as one string or per-entry mapping.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
    """Return the JSON schema for this dataset type.

    Args:
        pretty: Whether to format the schema output for readability.

    Returns:
        str | dict[str, str]: JSON schema as one string or per-entry mapping.
    """
    ...

update

update(m: SupportsKeysAndGetItem[_KT, _VT]) -> None

update(m: SupportsKeysAndGetItem[str, _VT], /, **kwargs: _VT) -> None

update(m: Iterable[tuple[_KT, _VT]]) -> None

update(m: Iterable[tuple[str, _VT]], /, **kwargs: _VT) -> None

update(**kwargs: _VT) -> None

D.update([E, ]**F) -> None. Update D from mapping/iterable E and F. If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

Source code in src/omnipy/shared/protocols/typing.py

def update(
    self,
    m: (SupportsKeysAndGetItem[_KT, _VT] | SupportsKeysAndGetItem[str, _VT]
        | Iterable[tuple[_KT, _VT]] | Iterable[tuple[str, _VT]] | None) = None,
    /,
    **kwargs: _VT,
) -> None:
    """
    D.update([E, ]**F) -> None.  Update D from mapping/iterable E and F.
    If E present and has a .keys() method, does:     for k in E: D[k] = E[k]
    If E present and lacks .keys() method, does:     for (k, v) in E: D[k] = v
    In either case, this is followed by: for k, v in F.items(): D[k] = v
    """
    raise AssumedToBeImplementedException

values

values() -> IsValuesView[_VT_co]

D.values() -> an object providing a view on D's values

Source code in src/omnipy/shared/protocols/typing.py

def values(self) -> IsValuesView[_VT_co]:
    """
    D.values() -> an object providing a view on D's values
    """
    raise AssumedToBeImplementedException

IsFailedData dataclass

Bases: Protocol

              flowchart BT
              omnipy.shared.protocols.data.IsFailedData[IsFailedData]

              

              click omnipy.shared.protocols.data.IsFailedData href "" "omnipy.shared.protocols.data.IsFailedData"
            

Metadata describing a dataset entry whose asynchronous load failed.

Returned by:

• Reference Code reference omnipy
  • data
    •  dataset  Dataset  failed_task_details
    •  multi  MultiModelDataset  failed_task_details
  • shared protocols  data
    •  HasData  data
    •  IsDataset  failed_task_details
    •  IsHttpUrlDataset  failed_task_details
    •  IsMultiModelDataset  failed_task_details

METHOD	DESCRIPTION
__init__

ATTRIBUTE	DESCRIPTION
exception	TYPE: BaseException
job_name	TYPE: str
job_unique_name	TYPE: str

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
@dataclass(frozen=True, kw_only=True)
class IsFailedData(Protocol):
    """Metadata describing a dataset entry whose asynchronous load failed."""

    job_name: str
    job_unique_name: str
    exception: BaseException

exception instance-attribute

exception: BaseException

job_name instance-attribute

job_name: str

job_unique_name instance-attribute

job_unique_name: str

__init__

__init__(*, job_name: str, job_unique_name: str, exception: BaseException) -> None

IsHttpUrlDataset

Bases: IsDataset, Protocol

              flowchart BT
              omnipy.shared.protocols.data.IsHttpUrlDataset[IsHttpUrlDataset]
              omnipy.shared.protocols.data.IsDataset[IsDataset]
              omnipy.shared.protocols.typing.IsMutableMapping[IsMutableMapping]
              omnipy.shared.protocols.typing.IsMapping[IsMapping]

                              omnipy.shared.protocols.data.IsDataset --> omnipy.shared.protocols.data.IsHttpUrlDataset
                                omnipy.shared.protocols.typing.IsMutableMapping --> omnipy.shared.protocols.data.IsDataset
                                omnipy.shared.protocols.typing.IsMapping --> omnipy.shared.protocols.typing.IsMutableMapping
                




              click omnipy.shared.protocols.data.IsHttpUrlDataset href "" "omnipy.shared.protocols.data.IsHttpUrlDataset"
              click omnipy.shared.protocols.data.IsDataset href "" "omnipy.shared.protocols.data.IsDataset"
              click omnipy.shared.protocols.typing.IsMutableMapping href "" "omnipy.shared.protocols.typing.IsMutableMapping"
              click omnipy.shared.protocols.typing.IsMapping href "" "omnipy.shared.protocols.typing.IsMapping"
            

Dataset protocol for collections of HTTP URL model entries.

METHOD	DESCRIPTION
__init__
clear	D.clear() -> None. Remove all items from D.
failed_task_details	Return failure metadata for entries whose processing failed.
from_data	Populate the dataset from plain Python data.
from_json	Populate the dataset from JSON-encoded entries.
get	D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.
get_type	Return the item type stored by this dataset class.
keys	D.keys() -> a set-like object providing a view on D's keys
load	Load dataset content from one or more paths or URLs.
load_into	Load external content into the current dataset instance.
pending_task_details	Return metadata for entries that are still processing.
pop	D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
popitem	D.popitem() -> (k, v), remove and return some (key, value) pair
save	Persist the dataset to a filesystem path.
setdefault	D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
to_data	Convert the dataset to plain Python data.
to_json	Serialize the dataset to JSON strings.
to_json_schema	Return the JSON schema for this dataset type.
update	D.update([E, ]**F) -> None. Update D from mapping/iterable E and F.
values	D.values() -> an object providing a view on D's values

ATTRIBUTE	DESCRIPTION
available_data	Return a view containing only successfully available entries. TYPE: Self
failed_data	Return a view containing entries whose processing failed. TYPE: Self
pending_data	Return a view containing entries that are still processing. TYPE: Self

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsHttpUrlDataset(IsDataset, Protocol):
    """Dataset protocol for collections of HTTP URL model entries."""

    ...

available_data property

available_data: Self

Return a view containing only successfully available entries.

RETURNS	DESCRIPTION
Self	Dataset containing entries whose content is already available. TYPE: Self

failed_data property

failed_data: Self

Return a view containing entries whose processing failed.

RETURNS	DESCRIPTION
Self	Dataset containing entries associated with failures. TYPE: Self

pending_data property

pending_data: Self

Return a view containing entries that are still processing.

RETURNS	DESCRIPTION
Self	Dataset containing entries backed by pending work. TYPE: Self

__init__

__init__(
    value: Mapping[str, object] | Iterator[tuple[str, object]] | UndefinedType = Undefined,
    *,
    data: Mapping[str, object] | UndefinedType = Undefined,
    **input_data: object,
) -> None

Source code in src/omnipy/shared/protocols/data.py

def __init__(
    self,
    value: Mapping[str, object] | Iterator[tuple[str, object]] | UndefinedType = Undefined,
    *,
    data: Mapping[str, object] | UndefinedType = Undefined,
    **input_data: object,
) -> None:
    ...

clear

clear() -> None

D.clear() -> None. Remove all items from D.

Source code in src/omnipy/shared/protocols/typing.py

def clear(self) -> None:
    """
    D.clear() -> None.  Remove all items from D.
    """
    raise AssumedToBeImplementedException

failed_task_details

failed_task_details() -> dict[str, IsFailedData]

Return failure metadata for entries whose processing failed.

RETURNS	DESCRIPTION
dict[str, IsFailedData]	dict[str, IsFailedData]: Failure metadata keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def failed_task_details(self) -> dict[str, IsFailedData]:
    """Return failure metadata for entries whose processing failed.

    Returns:
        dict[str, IsFailedData]: Failure metadata keyed by dataset entry name.
    """
    ...

from_data

from_data(data: Mapping[str, Any] | Iterable[tuple[str, Any]], update: bool = True) -> None

Populate the dataset from plain Python data.

PARAMETER	DESCRIPTION
data	Plain-data mapping or iterable of key-value pairs to import. TYPE: Mapping[str, Any] | Iterable[tuple[str, Any]]
update	Whether imported values should merge into existing content. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/data.py

def from_data(self,
              data: Mapping[str, Any] | Iterable[tuple[str, Any]],
              update: bool = True) -> None:
    """Populate the dataset from plain Python data.

    Args:
        data: Plain-data mapping or iterable of key-value pairs to import.
        update: Whether imported values should merge into existing content.
    """
    ...

from_json

from_json(data: Mapping[str, str] | Iterable[tuple[str, str]], update: bool = True) -> None

Populate the dataset from JSON-encoded entries.

PARAMETER	DESCRIPTION
data	JSON strings keyed by dataset entry name. TYPE: Mapping[str, str] | Iterable[tuple[str, str]]
update	Whether imported values should merge into existing content. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/data.py

def from_json(self,
              data: Mapping[str, str] | Iterable[tuple[str, str]],
              update: bool = True) -> None:
    """Populate the dataset from JSON-encoded entries.

    Args:
        data: JSON strings keyed by dataset entry name.
        update: Whether imported values should merge into existing content.
    """
    ...

get

get(key: _KT) -> _VT_co | None

get(key: _KT, default: _VT_co) -> _VT_co

get(key: _KT, default: _T) -> _VT_co | _T

D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.

Source code in src/omnipy/shared/protocols/typing.py

def get(self, key: _KT, default: None | _VT_co | _T = None, /) -> _VT_co | _T | None:
    """
    D.get(k[,d]) -> D[k] if k in D, else d.  d defaults to None.
    """
    raise AssumedToBeImplementedException

get_type cached classmethod

get_type() -> type[_ModelOrDatasetT]

Return the item type stored by this dataset class.

RETURNS	DESCRIPTION
type[_ModelOrDatasetT]	type[_ModelOrDatasetT]: Model or nested-dataset type used for items.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
@functools.cache
def get_type(cls) -> type[_ModelOrDatasetT]:
    """Return the item type stored by this dataset class.

    Returns:
        type[_ModelOrDatasetT]: Model or nested-dataset type used for items.
    """
    ...

keys

keys() -> IsKeysView[_KT]

D.keys() -> a set-like object providing a view on D's keys

Source code in src/omnipy/shared/protocols/typing.py

def keys(self) -> IsKeysView[_KT]:
    """
    D.keys() -> a set-like object providing a view on D's keys
    """
    raise AssumedToBeImplementedException

load classmethod

load(
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]

Load dataset content from one or more paths or URLs.

PARAMETER	DESCRIPTION
paths_or_urls	Source path, URL, or collection of sources to load. TYPE: IsPathsOrUrlsOneOrMoreOrNone DEFAULT: None
by_file_suffix	Whether serializer lookup should prefer file suffixes. TYPE: bool DEFAULT: False
as_mime_type	Explicit MIME type override, if any. TYPE: None | str DEFAULT: None
kwargs	Additional named path or URL sources. TYPE: IsPathOrUrl DEFAULT: {}

RETURNS	DESCRIPTION
Self | asyncio.Task[Self]	Self | asyncio.Task[Self]: Loaded dataset or asynchronous load task.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def load(
    cls,
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]:
    """Load dataset content from one or more paths or URLs.

    Args:
        paths_or_urls: Source path, URL, or collection of sources to load.
        by_file_suffix: Whether serializer lookup should prefer file suffixes.
        as_mime_type: Explicit MIME type override, if any.
        kwargs: Additional named path or URL sources.

    Returns:
        Self | asyncio.Task[Self]: Loaded dataset or asynchronous load task.
    """
    ...

load_into

load_into(
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]

Load external content into the current dataset instance.

PARAMETER	DESCRIPTION
paths_or_urls	Source path, URL, or collection of sources to load. TYPE: IsPathsOrUrlsOneOrMoreOrNone DEFAULT: None
by_file_suffix	Whether serializer lookup should prefer file suffixes. TYPE: bool DEFAULT: False
as_mime_type	Explicit MIME type override, if any. TYPE: None | str DEFAULT: None
kwargs	Additional named path or URL sources. TYPE: IsPathOrUrl DEFAULT: {}

RETURNS	DESCRIPTION
Self | asyncio.Task[Self]	Self | asyncio.Task[Self]: Updated dataset or asynchronous load task.

Source code in src/omnipy/shared/protocols/data.py

def load_into(
    self,
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]:
    """Load external content into the current dataset instance.

    Args:
        paths_or_urls: Source path, URL, or collection of sources to load.
        by_file_suffix: Whether serializer lookup should prefer file suffixes.
        as_mime_type: Explicit MIME type override, if any.
        kwargs: Additional named path or URL sources.

    Returns:
        Self | asyncio.Task[Self]: Updated dataset or asynchronous load task.
    """
    ...

pending_task_details

pending_task_details() -> dict[str, IsPendingData]

Return metadata for entries that are still processing.

RETURNS	DESCRIPTION
dict[str, IsPendingData]	dict[str, IsPendingData]: Pending metadata keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def pending_task_details(self) -> dict[str, IsPendingData]:
    """Return metadata for entries that are still processing.

    Returns:
        dict[str, IsPendingData]: Pending metadata keyed by dataset entry name.
    """
    ...

pop

pop(key: _KT) -> _VT

pop(key: _KT, default: _VT) -> _VT

pop(key: _KT, default: _T) -> _VT | _T

D.pop(k[,d]) -> v, remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised.

Source code in src/omnipy/shared/protocols/typing.py

def pop(self, key: _KT, default: None | _VT | _T = None, /) -> _VT | _T | None:
    """
    D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
    If key is not found, d is returned if given, otherwise KeyError is raised.
    """
    raise AssumedToBeImplementedException

popitem

popitem() -> tuple[_KT, _VT]

D.popitem() -> (k, v), remove and return some (key, value) pair as a 2-tuple; but raise KeyError if D is empty.

Source code in src/omnipy/shared/protocols/typing.py

def popitem(self) -> tuple[_KT, _VT]:
    """
    D.popitem() -> (k, v), remove and return some (key, value) pair
       as a 2-tuple; but raise KeyError if D is empty.
    """
    raise AssumedToBeImplementedException

save

save(path: str) -> None

Persist the dataset to a filesystem path.

PARAMETER	DESCRIPTION
path	Destination path for the serialized dataset. TYPE: str

Source code in src/omnipy/shared/protocols/data.py

def save(self, path: str) -> None:
    """Persist the dataset to a filesystem path.

    Args:
        path: Destination path for the serialized dataset.
    """
    ...

setdefault

setdefault(key: _KT, default: _VT) -> _VT

D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D

Source code in src/omnipy/shared/protocols/typing.py

def setdefault(self, key: _KT, default: _VT, /) -> _VT:
    """
    D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
    """
    raise AssumedToBeImplementedException

to_data

to_data() -> dict[str, Any]

Convert the dataset to plain Python data.

RETURNS	DESCRIPTION
dict[str, Any]	dict[str, Any]: Plain-data representation keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def to_data(self) -> dict[str, Any]:
    """Convert the dataset to plain Python data.

    Returns:
        dict[str, Any]: Plain-data representation keyed by dataset entry name.
    """
    ...

to_json

to_json(pretty=True) -> dict[str, str]

Serialize the dataset to JSON strings.

PARAMETER	DESCRIPTION
pretty	Whether to format the JSON output for readability. DEFAULT: True

RETURNS	DESCRIPTION
dict[str, str]	dict[str, str]: JSON representation for each dataset entry.

Source code in src/omnipy/shared/protocols/data.py

def to_json(self, pretty=True) -> dict[str, str]:
    """Serialize the dataset to JSON strings.

    Args:
        pretty: Whether to format the JSON output for readability.

    Returns:
        dict[str, str]: JSON representation for each dataset entry.
    """
    ...

to_json_schema classmethod

to_json_schema(pretty=True) -> str | dict[str, str]

Return the JSON schema for this dataset type.

PARAMETER	DESCRIPTION
pretty	Whether to format the schema output for readability. DEFAULT: True

RETURNS	DESCRIPTION
str | dict[str, str]	str | dict[str, str]: JSON schema as one string or per-entry mapping.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
    """Return the JSON schema for this dataset type.

    Args:
        pretty: Whether to format the schema output for readability.

    Returns:
        str | dict[str, str]: JSON schema as one string or per-entry mapping.
    """
    ...

update

update(m: SupportsKeysAndGetItem[_KT, _VT]) -> None

update(m: SupportsKeysAndGetItem[str, _VT], /, **kwargs: _VT) -> None

update(m: Iterable[tuple[_KT, _VT]]) -> None

update(m: Iterable[tuple[str, _VT]], /, **kwargs: _VT) -> None

update(**kwargs: _VT) -> None

D.update([E, ]**F) -> None. Update D from mapping/iterable E and F. If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

Source code in src/omnipy/shared/protocols/typing.py

def update(
    self,
    m: (SupportsKeysAndGetItem[_KT, _VT] | SupportsKeysAndGetItem[str, _VT]
        | Iterable[tuple[_KT, _VT]] | Iterable[tuple[str, _VT]] | None) = None,
    /,
    **kwargs: _VT,
) -> None:
    """
    D.update([E, ]**F) -> None.  Update D from mapping/iterable E and F.
    If E present and has a .keys() method, does:     for k in E: D[k] = E[k]
    If E present and lacks .keys() method, does:     for (k, v) in E: D[k] = v
    In either case, this is followed by: for k, v in F.items(): D[k] = v
    """
    raise AssumedToBeImplementedException

values

values() -> IsValuesView[_VT_co]

D.values() -> an object providing a view on D's values

Source code in src/omnipy/shared/protocols/typing.py

def values(self) -> IsValuesView[_VT_co]:
    """
    D.values() -> an object providing a view on D's values
    """
    raise AssumedToBeImplementedException

IsHttpUrlModel

Bases: IsModel, Protocol

              flowchart BT
              omnipy.shared.protocols.data.IsHttpUrlModel[IsHttpUrlModel]
              omnipy.shared.protocols.data.IsModel[IsModel]
              omnipy.shared.protocols.data.HasContent[HasContent]

                              omnipy.shared.protocols.data.IsModel --> omnipy.shared.protocols.data.IsHttpUrlModel
                                omnipy.shared.protocols.data.HasContent --> omnipy.shared.protocols.data.IsModel
                



              click omnipy.shared.protocols.data.IsHttpUrlModel href "" "omnipy.shared.protocols.data.IsHttpUrlModel"
              click omnipy.shared.protocols.data.IsModel href "" "omnipy.shared.protocols.data.IsModel"
              click omnipy.shared.protocols.data.HasContent href "" "omnipy.shared.protocols.data.HasContent"
            

Model protocol representing a validated HTTP URL value.

METHOD	DESCRIPTION
full_type	Return the fully resolved Python type represented by this model.

ATTRIBUTE	DESCRIPTION
content	Return the wrapped content value. TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsHttpUrlModel(IsModel, Protocol):
    """Model protocol representing a validated HTTP URL value."""

    ...

content property writable

content: ContentT

Return the wrapped content value.

RETURNS	DESCRIPTION
ContentT	Current content stored by the object. TYPE: ContentT

full_type classmethod

full_type() -> type[_RootT]

Return the fully resolved Python type represented by this model.

RETURNS	DESCRIPTION
type[_RootT]	type[_RootT]: Concrete Python type accepted as model content.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def full_type(cls) -> type[_RootT]:
    """Return the fully resolved Python type represented by this model.

    Returns:
        type[_RootT]: Concrete Python type accepted as model content.
    """
    ...

IsModel

Bases: HasContent[_RootT], Protocol[_RootT]

              flowchart BT
              omnipy.shared.protocols.data.IsModel[IsModel]
              omnipy.shared.protocols.data.HasContent[HasContent]

                              omnipy.shared.protocols.data.HasContent --> omnipy.shared.protocols.data.IsModel
                


              click omnipy.shared.protocols.data.IsModel href "" "omnipy.shared.protocols.data.IsModel"
              click omnipy.shared.protocols.data.HasContent href "" "omnipy.shared.protocols.data.HasContent"
            

Single-value data wrapper with typed content and conversion support.

Returned by:

• Reference Code reference omnipy shared protocols
  •  config
    •  IsBrowserUserInterfaceConfig  as_model
    •  IsColorConfig  as_model
    •  IsConfigBase  as_model
    •  IsDataConfig  as_model
    •  IsDimsModeConfig  as_model
    •  IsEngineConfig  as_model
    •  IsFontConfig  as_model
    •  IsHtmlUserInterfaceConfig  as_model
    •  IsHttpConfig  as_model
    •  IsHttpRequestsConfig  as_model
    •  IsJobConfig  as_model
    •  IsJobRunnerConfig  as_model
    •  IsJupyterUserInterfaceConfig  as_model
    •  IsLayoutConfig  as_model
    •  IsLocalOutputStorageConfig  as_model
    •  IsLocalRunnerConfig  as_model
    •  IsModelConfig  as_model
    •  IsOutputStorageConfig  as_model
    •  IsOutputStorageConfigBase  as_model
    •  IsOverflowConfig  as_model
    •  IsPrefectEngineConfig  as_model
    •  IsRootLogConfig  as_model
    •  IsS3OutputStorageConfig  as_model
    •  IsTerminalUserInterfaceConfig  as_model
    •  IsTextConfig  as_model
    •  IsUserInterfaceConfig  as_model
    •  IsUserInterfaceTypeConfig  as_model
  •  data  IsMultiModelDataset  get_model
  • hub  runtime  IsRuntimeConfig  as_model

Subclassed by:

• Reference Code reference omnipy shared protocols  data  IsHttpUrlModel

Used by:

• Reference Code reference omnipy shared protocols  data  IsMultiModelDataset  set_model

METHOD	DESCRIPTION
full_type	Return the fully resolved Python type represented by this model.

ATTRIBUTE	DESCRIPTION
content	Return the wrapped content value. TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsModel(HasContent[_RootT], Protocol[_RootT]):
    """Single-value data wrapper with typed content and conversion support."""
    @classmethod
    def full_type(cls) -> type[_RootT]:
        """Return the fully resolved Python type represented by this model.

        Returns:
            type[_RootT]: Concrete Python type accepted as model content.
        """
        ...

content property writable

content: ContentT

Return the wrapped content value.

RETURNS	DESCRIPTION
ContentT	Current content stored by the object. TYPE: ContentT

full_type classmethod

full_type() -> type[_RootT]

Return the fully resolved Python type represented by this model.

RETURNS	DESCRIPTION
type[_RootT]	type[_RootT]: Concrete Python type accepted as model content.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def full_type(cls) -> type[_RootT]:
    """Return the fully resolved Python type represented by this model.

    Returns:
        type[_RootT]: Concrete Python type accepted as model content.
    """
    ...

IsMultiModelDataset

Bases: IsDataset[_ModelOrDatasetT], Protocol[_ModelOrDatasetT]

              flowchart BT
              omnipy.shared.protocols.data.IsMultiModelDataset[IsMultiModelDataset]
              omnipy.shared.protocols.data.IsDataset[IsDataset]
              omnipy.shared.protocols.typing.IsMutableMapping[IsMutableMapping]
              omnipy.shared.protocols.typing.IsMapping[IsMapping]

                              omnipy.shared.protocols.data.IsDataset --> omnipy.shared.protocols.data.IsMultiModelDataset
                                omnipy.shared.protocols.typing.IsMutableMapping --> omnipy.shared.protocols.data.IsDataset
                                omnipy.shared.protocols.typing.IsMapping --> omnipy.shared.protocols.typing.IsMutableMapping
                




              click omnipy.shared.protocols.data.IsMultiModelDataset href "" "omnipy.shared.protocols.data.IsMultiModelDataset"
              click omnipy.shared.protocols.data.IsDataset href "" "omnipy.shared.protocols.data.IsDataset"
              click omnipy.shared.protocols.typing.IsMutableMapping href "" "omnipy.shared.protocols.typing.IsMutableMapping"
              click omnipy.shared.protocols.typing.IsMapping href "" "omnipy.shared.protocols.typing.IsMapping"
            

Dataset protocol that can assign different model classes per item.

Returned by:

• Reference Code reference
  •  omnipy  Dataset  as_multi_model_dataset
  • omnipy data
    •  dataset  Dataset  as_multi_model_dataset
    •  multi  MultiModelDataset  as_multi_model_dataset

METHOD	DESCRIPTION
__init__
clear	D.clear() -> None. Remove all items from D.
failed_task_details	Return failure metadata for entries whose processing failed.
from_data	Populate the dataset from plain Python data.
from_json	Populate the dataset from JSON-encoded entries.
get	D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.
get_model	Return the model class associated with one dataset entry.
get_type	Return the item type stored by this dataset class.
keys	D.keys() -> a set-like object providing a view on D's keys
load	Load dataset content from one or more paths or URLs.
load_into	Load external content into the current dataset instance.
pending_task_details	Return metadata for entries that are still processing.
pop	D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
popitem	D.popitem() -> (k, v), remove and return some (key, value) pair
save	Persist the dataset to a filesystem path.
set_model	Assign a model class to one dataset entry.
setdefault	D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
to_data	Convert the dataset to plain Python data.
to_json	Serialize the dataset to JSON strings.
to_json_schema	Return the JSON schema for this dataset type.
update	D.update([E, ]**F) -> None. Update D from mapping/iterable E and F.
values	D.values() -> an object providing a view on D's values

ATTRIBUTE	DESCRIPTION
available_data	Return a view containing only successfully available entries. TYPE: Self
failed_data	Return a view containing entries whose processing failed. TYPE: Self
pending_data	Return a view containing entries that are still processing. TYPE: Self

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsMultiModelDataset(IsDataset[_ModelOrDatasetT], Protocol[_ModelOrDatasetT]):
    """Dataset protocol that can assign different model classes per item."""
    def set_model(self, data_file: str, model: type[IsModel]) -> None:
        """Assign a model class to one dataset entry.

        Args:
            data_file: Entry name whose model should be updated.
            model: Model class to associate with that entry.
        """
        ...

    def get_model(self, data_file: str) -> type[IsModel]:
        """Return the model class associated with one dataset entry.

        Args:
            data_file: Entry name whose model should be returned.

        Returns:
            type[IsModel]: Model class associated with that entry.
        """
        ...

available_data property

available_data: Self

Return a view containing only successfully available entries.

RETURNS	DESCRIPTION
Self	Dataset containing entries whose content is already available. TYPE: Self

failed_data property

failed_data: Self

Return a view containing entries whose processing failed.

RETURNS	DESCRIPTION
Self	Dataset containing entries associated with failures. TYPE: Self

pending_data property

pending_data: Self

Return a view containing entries that are still processing.

RETURNS	DESCRIPTION
Self	Dataset containing entries backed by pending work. TYPE: Self

__init__

__init__(
    value: Mapping[str, object] | Iterator[tuple[str, object]] | UndefinedType = Undefined,
    *,
    data: Mapping[str, object] | UndefinedType = Undefined,
    **input_data: object,
) -> None

Source code in src/omnipy/shared/protocols/data.py

def __init__(
    self,
    value: Mapping[str, object] | Iterator[tuple[str, object]] | UndefinedType = Undefined,
    *,
    data: Mapping[str, object] | UndefinedType = Undefined,
    **input_data: object,
) -> None:
    ...

clear

clear() -> None

D.clear() -> None. Remove all items from D.

Source code in src/omnipy/shared/protocols/typing.py

def clear(self) -> None:
    """
    D.clear() -> None.  Remove all items from D.
    """
    raise AssumedToBeImplementedException

failed_task_details

failed_task_details() -> dict[str, IsFailedData]

Return failure metadata for entries whose processing failed.

RETURNS	DESCRIPTION
dict[str, IsFailedData]	dict[str, IsFailedData]: Failure metadata keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def failed_task_details(self) -> dict[str, IsFailedData]:
    """Return failure metadata for entries whose processing failed.

    Returns:
        dict[str, IsFailedData]: Failure metadata keyed by dataset entry name.
    """
    ...

from_data

from_data(data: Mapping[str, Any] | Iterable[tuple[str, Any]], update: bool = True) -> None

Populate the dataset from plain Python data.

PARAMETER	DESCRIPTION
data	Plain-data mapping or iterable of key-value pairs to import. TYPE: Mapping[str, Any] | Iterable[tuple[str, Any]]
update	Whether imported values should merge into existing content. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/data.py

def from_data(self,
              data: Mapping[str, Any] | Iterable[tuple[str, Any]],
              update: bool = True) -> None:
    """Populate the dataset from plain Python data.

    Args:
        data: Plain-data mapping or iterable of key-value pairs to import.
        update: Whether imported values should merge into existing content.
    """
    ...

from_json

from_json(data: Mapping[str, str] | Iterable[tuple[str, str]], update: bool = True) -> None

Populate the dataset from JSON-encoded entries.

PARAMETER	DESCRIPTION
data	JSON strings keyed by dataset entry name. TYPE: Mapping[str, str] | Iterable[tuple[str, str]]
update	Whether imported values should merge into existing content. TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/data.py

def from_json(self,
              data: Mapping[str, str] | Iterable[tuple[str, str]],
              update: bool = True) -> None:
    """Populate the dataset from JSON-encoded entries.

    Args:
        data: JSON strings keyed by dataset entry name.
        update: Whether imported values should merge into existing content.
    """
    ...

get

get(key: _KT) -> _VT_co | None

get(key: _KT, default: _VT_co) -> _VT_co

get(key: _KT, default: _T) -> _VT_co | _T

D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.

Source code in src/omnipy/shared/protocols/typing.py

def get(self, key: _KT, default: None | _VT_co | _T = None, /) -> _VT_co | _T | None:
    """
    D.get(k[,d]) -> D[k] if k in D, else d.  d defaults to None.
    """
    raise AssumedToBeImplementedException

get_model

get_model(data_file: str) -> type[IsModel]

Return the model class associated with one dataset entry.

PARAMETER	DESCRIPTION
data_file	Entry name whose model should be returned. TYPE: str

RETURNS	DESCRIPTION
type[IsModel]	type[IsModel]: Model class associated with that entry.

Source code in src/omnipy/shared/protocols/data.py

def get_model(self, data_file: str) -> type[IsModel]:
    """Return the model class associated with one dataset entry.

    Args:
        data_file: Entry name whose model should be returned.

    Returns:
        type[IsModel]: Model class associated with that entry.
    """
    ...

get_type cached classmethod

get_type() -> type[_ModelOrDatasetT]

Return the item type stored by this dataset class.

RETURNS	DESCRIPTION
type[_ModelOrDatasetT]	type[_ModelOrDatasetT]: Model or nested-dataset type used for items.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
@functools.cache
def get_type(cls) -> type[_ModelOrDatasetT]:
    """Return the item type stored by this dataset class.

    Returns:
        type[_ModelOrDatasetT]: Model or nested-dataset type used for items.
    """
    ...

keys

keys() -> IsKeysView[_KT]

D.keys() -> a set-like object providing a view on D's keys

Source code in src/omnipy/shared/protocols/typing.py

def keys(self) -> IsKeysView[_KT]:
    """
    D.keys() -> a set-like object providing a view on D's keys
    """
    raise AssumedToBeImplementedException

load classmethod

load(
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]

Load dataset content from one or more paths or URLs.

PARAMETER	DESCRIPTION
paths_or_urls	Source path, URL, or collection of sources to load. TYPE: IsPathsOrUrlsOneOrMoreOrNone DEFAULT: None
by_file_suffix	Whether serializer lookup should prefer file suffixes. TYPE: bool DEFAULT: False
as_mime_type	Explicit MIME type override, if any. TYPE: None | str DEFAULT: None
kwargs	Additional named path or URL sources. TYPE: IsPathOrUrl DEFAULT: {}

RETURNS	DESCRIPTION
Self | asyncio.Task[Self]	Self | asyncio.Task[Self]: Loaded dataset or asynchronous load task.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def load(
    cls,
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]:
    """Load dataset content from one or more paths or URLs.

    Args:
        paths_or_urls: Source path, URL, or collection of sources to load.
        by_file_suffix: Whether serializer lookup should prefer file suffixes.
        as_mime_type: Explicit MIME type override, if any.
        kwargs: Additional named path or URL sources.

    Returns:
        Self | asyncio.Task[Self]: Loaded dataset or asynchronous load task.
    """
    ...

load_into

load_into(
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]

Load external content into the current dataset instance.

PARAMETER	DESCRIPTION
paths_or_urls	Source path, URL, or collection of sources to load. TYPE: IsPathsOrUrlsOneOrMoreOrNone DEFAULT: None
by_file_suffix	Whether serializer lookup should prefer file suffixes. TYPE: bool DEFAULT: False
as_mime_type	Explicit MIME type override, if any. TYPE: None | str DEFAULT: None
kwargs	Additional named path or URL sources. TYPE: IsPathOrUrl DEFAULT: {}

RETURNS	DESCRIPTION
Self | asyncio.Task[Self]	Self | asyncio.Task[Self]: Updated dataset or asynchronous load task.

Source code in src/omnipy/shared/protocols/data.py

def load_into(
    self,
    paths_or_urls: IsPathsOrUrlsOneOrMoreOrNone = None,
    by_file_suffix: bool = False,
    as_mime_type: None | str = None,
    **kwargs: IsPathOrUrl,
) -> Self | asyncio.Task[Self]:
    """Load external content into the current dataset instance.

    Args:
        paths_or_urls: Source path, URL, or collection of sources to load.
        by_file_suffix: Whether serializer lookup should prefer file suffixes.
        as_mime_type: Explicit MIME type override, if any.
        kwargs: Additional named path or URL sources.

    Returns:
        Self | asyncio.Task[Self]: Updated dataset or asynchronous load task.
    """
    ...

pending_task_details

pending_task_details() -> dict[str, IsPendingData]

Return metadata for entries that are still processing.

RETURNS	DESCRIPTION
dict[str, IsPendingData]	dict[str, IsPendingData]: Pending metadata keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def pending_task_details(self) -> dict[str, IsPendingData]:
    """Return metadata for entries that are still processing.

    Returns:
        dict[str, IsPendingData]: Pending metadata keyed by dataset entry name.
    """
    ...

pop

pop(key: _KT) -> _VT

pop(key: _KT, default: _VT) -> _VT

pop(key: _KT, default: _T) -> _VT | _T

D.pop(k[,d]) -> v, remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised.

Source code in src/omnipy/shared/protocols/typing.py

def pop(self, key: _KT, default: None | _VT | _T = None, /) -> _VT | _T | None:
    """
    D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
    If key is not found, d is returned if given, otherwise KeyError is raised.
    """
    raise AssumedToBeImplementedException

popitem

popitem() -> tuple[_KT, _VT]

D.popitem() -> (k, v), remove and return some (key, value) pair as a 2-tuple; but raise KeyError if D is empty.

Source code in src/omnipy/shared/protocols/typing.py

def popitem(self) -> tuple[_KT, _VT]:
    """
    D.popitem() -> (k, v), remove and return some (key, value) pair
       as a 2-tuple; but raise KeyError if D is empty.
    """
    raise AssumedToBeImplementedException

save

save(path: str) -> None

Persist the dataset to a filesystem path.

PARAMETER	DESCRIPTION
path	Destination path for the serialized dataset. TYPE: str

Source code in src/omnipy/shared/protocols/data.py

def save(self, path: str) -> None:
    """Persist the dataset to a filesystem path.

    Args:
        path: Destination path for the serialized dataset.
    """
    ...

set_model

set_model(data_file: str, model: type[IsModel]) -> None

Assign a model class to one dataset entry.

PARAMETER	DESCRIPTION
data_file	Entry name whose model should be updated. TYPE: str
model	Model class to associate with that entry. TYPE: type[IsModel]

Source code in src/omnipy/shared/protocols/data.py

def set_model(self, data_file: str, model: type[IsModel]) -> None:
    """Assign a model class to one dataset entry.

    Args:
        data_file: Entry name whose model should be updated.
        model: Model class to associate with that entry.
    """
    ...

setdefault

setdefault(key: _KT, default: _VT) -> _VT

D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D

Source code in src/omnipy/shared/protocols/typing.py

def setdefault(self, key: _KT, default: _VT, /) -> _VT:
    """
    D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
    """
    raise AssumedToBeImplementedException

to_data

to_data() -> dict[str, Any]

Convert the dataset to plain Python data.

RETURNS	DESCRIPTION
dict[str, Any]	dict[str, Any]: Plain-data representation keyed by dataset entry name.

Source code in src/omnipy/shared/protocols/data.py

def to_data(self) -> dict[str, Any]:
    """Convert the dataset to plain Python data.

    Returns:
        dict[str, Any]: Plain-data representation keyed by dataset entry name.
    """
    ...

to_json

to_json(pretty=True) -> dict[str, str]

Serialize the dataset to JSON strings.

PARAMETER	DESCRIPTION
pretty	Whether to format the JSON output for readability. DEFAULT: True

RETURNS	DESCRIPTION
dict[str, str]	dict[str, str]: JSON representation for each dataset entry.

Source code in src/omnipy/shared/protocols/data.py

def to_json(self, pretty=True) -> dict[str, str]:
    """Serialize the dataset to JSON strings.

    Args:
        pretty: Whether to format the JSON output for readability.

    Returns:
        dict[str, str]: JSON representation for each dataset entry.
    """
    ...

to_json_schema classmethod

to_json_schema(pretty=True) -> str | dict[str, str]

Return the JSON schema for this dataset type.

PARAMETER	DESCRIPTION
pretty	Whether to format the schema output for readability. DEFAULT: True

RETURNS	DESCRIPTION
str | dict[str, str]	str | dict[str, str]: JSON schema as one string or per-entry mapping.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
    """Return the JSON schema for this dataset type.

    Args:
        pretty: Whether to format the schema output for readability.

    Returns:
        str | dict[str, str]: JSON schema as one string or per-entry mapping.
    """
    ...

update

update(m: SupportsKeysAndGetItem[_KT, _VT]) -> None

update(m: SupportsKeysAndGetItem[str, _VT], /, **kwargs: _VT) -> None

update(m: Iterable[tuple[_KT, _VT]]) -> None

update(m: Iterable[tuple[str, _VT]], /, **kwargs: _VT) -> None

update(**kwargs: _VT) -> None

D.update([E, ]**F) -> None. Update D from mapping/iterable E and F. If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

Source code in src/omnipy/shared/protocols/typing.py

def update(
    self,
    m: (SupportsKeysAndGetItem[_KT, _VT] | SupportsKeysAndGetItem[str, _VT]
        | Iterable[tuple[_KT, _VT]] | Iterable[tuple[str, _VT]] | None) = None,
    /,
    **kwargs: _VT,
) -> None:
    """
    D.update([E, ]**F) -> None.  Update D from mapping/iterable E and F.
    If E present and has a .keys() method, does:     for k in E: D[k] = E[k]
    If E present and lacks .keys() method, does:     for (k, v) in E: D[k] = v
    In either case, this is followed by: for k, v in F.items(): D[k] = v
    """
    raise AssumedToBeImplementedException

values

values() -> IsValuesView[_VT_co]

D.values() -> an object providing a view on D's values

Source code in src/omnipy/shared/protocols/typing.py

def values(self) -> IsValuesView[_VT_co]:
    """
    D.values() -> an object providing a view on D's values
    """
    raise AssumedToBeImplementedException

IsPendingData dataclass

Bases: Protocol

              flowchart BT
              omnipy.shared.protocols.data.IsPendingData[IsPendingData]

              

              click omnipy.shared.protocols.data.IsPendingData href "" "omnipy.shared.protocols.data.IsPendingData"
            

Metadata describing a dataset entry that is still processing.

Used for items backed by asynchronous work that has not completed yet.

Returned by:

• Reference Code reference omnipy
  • data
    •  dataset  Dataset  pending_task_details
    •  multi  MultiModelDataset  pending_task_details
  • shared protocols  data
    •  HasData  data
    •  IsDataset  pending_task_details
    •  IsHttpUrlDataset  pending_task_details
    •  IsMultiModelDataset  pending_task_details

METHOD	DESCRIPTION
__init__

ATTRIBUTE	DESCRIPTION
job_name	TYPE: str
job_unique_name	TYPE: str

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
@dataclass(frozen=True, kw_only=True)
class IsPendingData(Protocol):
    """Metadata describing a dataset entry that is still processing.

    Used for items backed by asynchronous work that has not completed yet.
    """

    job_name: str
    job_unique_name: str

job_name instance-attribute

job_name: str

job_unique_name instance-attribute

job_unique_name: str

__init__

__init__(*, job_name: str, job_unique_name: str) -> None

IsReactive

Bases: Protocol[ContentT]

              flowchart BT
              omnipy.shared.protocols.data.IsReactive[IsReactive]

              

              click omnipy.shared.protocols.data.IsReactive href "" "omnipy.shared.protocols.data.IsReactive"
            

Mutable reactive value wrapper used for config/state propagation.

Returned by:

• Reference Code reference omnipy shared protocols  data  IsReactiveObjects
  •  available_display_dims_in_px
  •  jupyter_ui_config
  •  layout_config
  •  obj_id_update_flags
  •  text_config

METHOD	DESCRIPTION
set	Replace the current reactive value.

ATTRIBUTE	DESCRIPTION
value	Return the current reactive value. TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsReactive(Protocol[ContentT]):
    """Mutable reactive value wrapper used for config/state propagation."""
    @property
    def value(self) -> ContentT:
        """Return the current reactive value.

        Returns:
            ContentT: Current value stored by the reactive wrapper.
        """
        ...

    def set(self, value: ContentT):
        """Replace the current reactive value.

        Args:
            value: New value to publish.
        """
        ...

value property

value: ContentT

Return the current reactive value.

RETURNS	DESCRIPTION
ContentT	Current value stored by the reactive wrapper. TYPE: ContentT

set

set(value: ContentT)

Replace the current reactive value.

PARAMETER	DESCRIPTION
value	New value to publish. TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py

def set(self, value: ContentT):
    """Replace the current reactive value.

    Args:
        value: New value to publish.
    """
    ...

IsReactiveObjects

Bases: Protocol

              flowchart BT
              omnipy.shared.protocols.data.IsReactiveObjects[IsReactiveObjects]

              

              click omnipy.shared.protocols.data.IsReactiveObjects href "" "omnipy.shared.protocols.data.IsReactiveObjects"
            

Bundle of reactive objects shared by display and configuration layers.

Returned by:

• Reference Code reference omnipy
  • data
    •  dataset  Dataset  reactive_objects
    •  model  Model  reactive_objects
    •  multi  MultiModelDataset  reactive_objects
  • hub  runtime  RuntimeObjects  reactive
  • shared protocols
    •  data  IsDataClassCreator  reactive_objects
    • hub  runtime  IsRuntimeObjects  reactive

Used by:

• Reference Code reference omnipy shared protocols  data  IsDataClassCreator  set_reactive_objects

ATTRIBUTE	DESCRIPTION
available_display_dims_in_px	TYPE: IsReactive[AvailableDisplayDims]
jupyter_ui_config	TYPE: IsReactive[IsJupyterUserInterfaceConfig]
layout_config	TYPE: IsReactive[IsLayoutConfig]
obj_id_update_flags	TYPE: IsReactive[dict[int, bool]]
text_config	TYPE: IsReactive[IsTextConfig]

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsReactiveObjects(Protocol):
    """Bundle of reactive objects shared by display and configuration layers."""

    jupyter_ui_config: IsReactive[IsJupyterUserInterfaceConfig]
    text_config: IsReactive[IsTextConfig]
    layout_config: IsReactive[IsLayoutConfig]
    available_display_dims_in_px: IsReactive[AvailableDisplayDims]
    obj_id_update_flags: IsReactive[dict[int, bool]]

    def __eq__(self, other) -> bool:
        ...

available_display_dims_in_px instance-attribute

available_display_dims_in_px: IsReactive[AvailableDisplayDims]

jupyter_ui_config instance-attribute

jupyter_ui_config: IsReactive[IsJupyterUserInterfaceConfig]

layout_config instance-attribute

layout_config: IsReactive[IsLayoutConfig]

obj_id_update_flags instance-attribute

obj_id_update_flags: IsReactive[dict[int, bool]]

text_config instance-attribute

text_config: IsReactive[IsTextConfig]

IsSerializer

Bases: Protocol[_DatasetT]

              flowchart BT
              omnipy.shared.protocols.data.IsSerializer[IsSerializer]

              

              click omnipy.shared.protocols.data.IsSerializer href "" "omnipy.shared.protocols.data.IsSerializer"
            

Serializer interface for converting datasets to and from bytes.

Returned by:

• Reference Code reference omnipy
  • data  serializer  SerializerRegistry
    •  serializers
    •  auto_detect
    •  auto_detect_tar_file_serializer
  • shared protocols  data  IsSerializerRegistry
    •  serializers
    •  auto_detect
    •  auto_detect_tar_file_serializer

Subclassed by:

• Reference Code reference omnipy shared protocols  data  IsTarFileSerializer

Used by:

• Reference Code reference omnipy
  • data  serializer  SerializerRegistry  register
  • shared protocols  data  IsSerializerRegistry  register

METHOD	DESCRIPTION
deserialize	Deserialize a bytes payload into a dataset instance.
get_dataset_cls_for_new	Return the dataset class this serializer creates when deserializing.
get_output_file_suffix	Return the default file suffix produced by this serializer.
is_dataset_directly_supported	Return whether the serializer can handle the dataset as-is.
serialize	Serialize a dataset into a bytes-like payload.

Source code in src/omnipy/shared/protocols/data.py

class IsSerializer(Protocol[_DatasetT]):
    """Serializer interface for converting datasets to and from bytes."""
    @classmethod
    def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
        """Return whether the serializer can handle the dataset as-is.

        Args:
            dataset: Dataset instance to check.

        Returns:
            bool: ``True`` when no dataset conversion is required before serialization.
        """
        ...

    @classmethod
    def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
        """Return the dataset class this serializer creates when deserializing.

        Returns:
            Type[IsDataset]: Dataset class produced for fresh deserialization targets.
        """
        ...

    @classmethod
    def get_output_file_suffix(cls) -> str:
        """Return the default file suffix produced by this serializer.

        Returns:
            str: File suffix used for serialized output files.
        """
        ...

    @classmethod
    def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
        """Serialize a dataset into a bytes-like payload.

        Args:
            dataset: Dataset instance to serialize.

        Returns:
            bytes | memoryview: Serialized dataset payload.
        """
        ...

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

        Args:
            serialized: Serialized dataset payload.
            any_file_suffix: Whether suffix validation should be relaxed.

        Returns:
            _DatasetT: Deserialized dataset instance.
        """
        ...

deserialize classmethod

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

Deserialize a bytes payload into a dataset instance.

PARAMETER	DESCRIPTION
serialized	Serialized dataset payload. TYPE: bytes
any_file_suffix	Whether suffix validation should be relaxed. DEFAULT: False

RETURNS	DESCRIPTION
_DatasetT	Deserialized dataset instance. TYPE: _DatasetT

Source code in src/omnipy/shared/protocols/data.py

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

    Args:
        serialized: Serialized dataset payload.
        any_file_suffix: Whether suffix validation should be relaxed.

    Returns:
        _DatasetT: Deserialized dataset instance.
    """
    ...

get_dataset_cls_for_new classmethod

get_dataset_cls_for_new() -> Type[IsDataset]

Return the dataset class this serializer creates when deserializing.

RETURNS	DESCRIPTION
Type[IsDataset]	Type[IsDataset]: Dataset class produced for fresh deserialization targets.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
    """Return the dataset class this serializer creates when deserializing.

    Returns:
        Type[IsDataset]: Dataset class produced for fresh deserialization targets.
    """
    ...

get_output_file_suffix classmethod

get_output_file_suffix() -> str

Return the default file suffix produced by this serializer.

RETURNS	DESCRIPTION
str	File suffix used for serialized output files. TYPE: str

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def get_output_file_suffix(cls) -> str:
    """Return the default file suffix produced by this serializer.

    Returns:
        str: File suffix used for serialized output files.
    """
    ...

is_dataset_directly_supported classmethod

is_dataset_directly_supported(dataset: IsDataset) -> bool

Return whether the serializer can handle the dataset as-is.

PARAMETER	DESCRIPTION
dataset	Dataset instance to check. TYPE: IsDataset

RETURNS	DESCRIPTION
bool	True when no dataset conversion is required before serialization. TYPE: bool

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    """Return whether the serializer can handle the dataset as-is.

    Args:
        dataset: Dataset instance to check.

    Returns:
        bool: ``True`` when no dataset conversion is required before serialization.
    """
    ...

serialize classmethod

serialize(dataset: _DatasetT) -> bytes | memoryview

Serialize a dataset into a bytes-like payload.

PARAMETER	DESCRIPTION
dataset	Dataset instance to serialize. TYPE: _DatasetT

RETURNS	DESCRIPTION
bytes | memoryview	bytes | memoryview: Serialized dataset payload.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
    """Serialize a dataset into a bytes-like payload.

    Args:
        dataset: Dataset instance to serialize.

    Returns:
        bytes | memoryview: Serialized dataset payload.
    """
    ...

IsSerializerRegistry

Bases: Protocol

              flowchart BT
              omnipy.shared.protocols.data.IsSerializerRegistry[IsSerializerRegistry]

              

              click omnipy.shared.protocols.data.IsSerializerRegistry href "" "omnipy.shared.protocols.data.IsSerializerRegistry"
            

Registry that tracks serializers and selects suitable ones for datasets.

Returned by:

• Reference Code reference omnipy
  • hub  runtime  RuntimeObjects  serializers
  • shared protocols hub  runtime  IsRuntimeObjects  serializers

Used by:

• Reference Code reference omnipy  components  register_serializers

METHOD	DESCRIPTION
__init__
auto_detect	Return the best serializer match for a dataset, if any.
auto_detect_tar_file_serializer	Return the best tar-file serializer match for a dataset, if any.
detect_tar_file_serializers_from_dataset_cls	Return tar-file serializers compatible with the dataset class.
detect_tar_file_serializers_from_file_suffix	Return tar-file serializers matching a file suffix.
load_from_tar_file_path_based_on_dataset_cls	Load a tar archive into a dataset using dataset-class detection.
load_from_tar_file_path_based_on_file_suffix	Load a tar archive into a dataset using suffix-based detection.
register	Register a serializer class with the registry.

ATTRIBUTE	DESCRIPTION
serializers	Return all registered serializer classes. TYPE: tuple[Type[IsSerializer], ...]
tar_file_serializers	Return the registered tar-file serializer classes. TYPE: tuple[Type[IsTarFileSerializer], ...]

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsSerializerRegistry(Protocol):
    """Registry that tracks serializers and selects suitable ones for datasets."""
    def __init__(self) -> None:
        ...

    def register(self, serializer_cls: Type[IsSerializer]) -> None:
        """Register a serializer class with the registry.

        Args:
            serializer_cls: Serializer class to add.
        """
        ...

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

        Returns:
            tuple[Type[IsSerializer], ...]: Registered serializer classes.
        """
        ...

    @property
    def tar_file_serializers(self) -> tuple[Type[IsTarFileSerializer], ...]:
        """Return the registered tar-file serializer classes.

        Returns:
            tuple[Type[IsTarFileSerializer], ...]: Registered tar-file serializers.
        """
        ...

    def auto_detect(self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
        """Return the best serializer match for a dataset, if any.

        Args:
            dataset: Dataset to inspect.

        Returns:
            tuple[IsDataset, IsSerializer] | tuple[None, None]: Parsed dataset and
                serializer pair, or ``(None, None)`` when no match exists.
        """
        ...

    def auto_detect_tar_file_serializer(
            self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
        """Return the best tar-file serializer match for a dataset, if any.

        Args:
            dataset: Dataset to inspect.

        Returns:
            tuple[IsDataset, IsSerializer] | tuple[None, None]: Parsed dataset and
                tar-file serializer pair, or ``(None, None)`` when no match exists.
        """
        ...

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

    def detect_tar_file_serializers_from_dataset_cls(
            self, dataset: IsDataset) -> tuple[Type[IsTarFileSerializer], ...]:
        """Return tar-file serializers compatible with the dataset class.

        Args:
            dataset: Dataset whose class should be inspected.

        Returns:
            tuple[Type[IsTarFileSerializer], ...]: Compatible tar-file serializer classes.
        """
        ...

    def detect_tar_file_serializers_from_file_suffix(
            self, file_suffix: str) -> tuple[Type[IsTarFileSerializer], ...]:
        """Return tar-file serializers matching a file suffix.

        Args:
            file_suffix: File suffix to match against registered serializers.

        Returns:
            tuple[Type[IsTarFileSerializer], ...]: Matching tar-file serializer classes.
        """
        ...

    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 into a dataset using suffix-based detection.

        Args:
            log_obj: Logger used for progress and error reporting.
            tar_file_path: Path to the tar archive on disk.
            to_dataset: Dataset instance to populate.

        Returns:
            IsDataset | None: Populated dataset, or ``None`` when no serializer matches.
        """
        ...

    def load_from_tar_file_path_based_on_dataset_cls(self,
                                                     log_obj: CanLog,
                                                     tar_file_path: str,
                                                     to_dataset: IsDataset) -> IsDataset | None:
        """Load a tar archive into a dataset using dataset-class detection.

        Args:
            log_obj: Logger used for progress and error reporting.
            tar_file_path: Path to the tar archive on disk.
            to_dataset: Dataset instance to populate.

        Returns:
            IsDataset | None: Populated dataset, or ``None`` when no serializer matches.
        """
        ...

serializers property

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

Return all registered serializer classes.

RETURNS	DESCRIPTION
tuple[Type[IsSerializer], ...]	tuple[Type[IsSerializer], ...]: Registered serializer classes.

tar_file_serializers property

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

Return the registered tar-file serializer classes.

RETURNS	DESCRIPTION
tuple[Type[IsTarFileSerializer], ...]	tuple[Type[IsTarFileSerializer], ...]: Registered tar-file serializers.

__init__

__init__() -> None

Source code in src/omnipy/shared/protocols/data.py

def __init__(self) -> None:
    ...

auto_detect

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

Return the best serializer match for a dataset, if any.

PARAMETER	DESCRIPTION
dataset	Dataset to inspect. TYPE: IsDataset

RETURNS	DESCRIPTION
tuple[IsDataset, IsSerializer] | tuple[None, None]	tuple[IsDataset, IsSerializer] | tuple[None, None]: Parsed dataset and serializer pair, or (None, None) when no match exists.

Source code in src/omnipy/shared/protocols/data.py

def auto_detect(self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
    """Return the best serializer match for a dataset, if any.

    Args:
        dataset: Dataset to inspect.

    Returns:
        tuple[IsDataset, IsSerializer] | tuple[None, None]: Parsed dataset and
            serializer pair, or ``(None, None)`` when no match exists.
    """
    ...

auto_detect_tar_file_serializer

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

Return the best tar-file serializer match for a dataset, if any.

PARAMETER	DESCRIPTION
dataset	Dataset to inspect. TYPE: IsDataset

RETURNS	DESCRIPTION
tuple[IsDataset, IsSerializer] | tuple[None, None]	tuple[IsDataset, IsSerializer] | tuple[None, None]: Parsed dataset and tar-file serializer pair, or (None, None) when no match exists.

Source code in src/omnipy/shared/protocols/data.py

def auto_detect_tar_file_serializer(
        self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
    """Return the best tar-file serializer match for a dataset, if any.

    Args:
        dataset: Dataset to inspect.

    Returns:
        tuple[IsDataset, IsSerializer] | tuple[None, None]: Parsed dataset and
            tar-file serializer pair, or ``(None, None)`` when no match exists.
    """
    ...

detect_tar_file_serializers_from_dataset_cls

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

Return tar-file serializers compatible with the dataset class.

PARAMETER	DESCRIPTION
dataset	Dataset whose class should be inspected. TYPE: IsDataset

RETURNS	DESCRIPTION
tuple[Type[IsTarFileSerializer], ...]	tuple[Type[IsTarFileSerializer], ...]: Compatible tar-file serializer classes.

Source code in src/omnipy/shared/protocols/data.py

def detect_tar_file_serializers_from_dataset_cls(
        self, dataset: IsDataset) -> tuple[Type[IsTarFileSerializer], ...]:
    """Return tar-file serializers compatible with the dataset class.

    Args:
        dataset: Dataset whose class should be inspected.

    Returns:
        tuple[Type[IsTarFileSerializer], ...]: Compatible tar-file serializer classes.
    """
    ...

detect_tar_file_serializers_from_file_suffix

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

Return tar-file serializers matching a file suffix.

PARAMETER	DESCRIPTION
file_suffix	File suffix to match against registered serializers. TYPE: str

RETURNS	DESCRIPTION
tuple[Type[IsTarFileSerializer], ...]	tuple[Type[IsTarFileSerializer], ...]: Matching tar-file serializer classes.

Source code in src/omnipy/shared/protocols/data.py

def detect_tar_file_serializers_from_file_suffix(
        self, file_suffix: str) -> tuple[Type[IsTarFileSerializer], ...]:
    """Return tar-file serializers matching a file suffix.

    Args:
        file_suffix: File suffix to match against registered serializers.

    Returns:
        tuple[Type[IsTarFileSerializer], ...]: Matching tar-file serializer classes.
    """
    ...

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
) -> IsDataset | None

Load a tar archive into a dataset using dataset-class detection.

PARAMETER	DESCRIPTION
log_obj	Logger used for progress and error reporting. TYPE: CanLog
tar_file_path	Path to the tar archive on disk. TYPE: str
to_dataset	Dataset instance to populate. TYPE: IsDataset

RETURNS	DESCRIPTION
IsDataset | None	IsDataset | None: Populated dataset, or None when no serializer matches.

Source code in src/omnipy/shared/protocols/data.py

def load_from_tar_file_path_based_on_dataset_cls(self,
                                                 log_obj: CanLog,
                                                 tar_file_path: str,
                                                 to_dataset: IsDataset) -> IsDataset | None:
    """Load a tar archive into a dataset using dataset-class detection.

    Args:
        log_obj: Logger used for progress and error reporting.
        tar_file_path: Path to the tar archive on disk.
        to_dataset: Dataset instance to populate.

    Returns:
        IsDataset | None: Populated dataset, or ``None`` when no serializer matches.
    """
    ...

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 into a dataset using suffix-based detection.

PARAMETER	DESCRIPTION
log_obj	Logger used for progress and error reporting. TYPE: CanLog
tar_file_path	Path to the tar archive on disk. TYPE: str
to_dataset	Dataset instance to populate. TYPE: IsDataset

RETURNS	DESCRIPTION
IsDataset | None	IsDataset | None: Populated dataset, or None when no serializer matches.

Source code in src/omnipy/shared/protocols/data.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 into a dataset using suffix-based detection.

    Args:
        log_obj: Logger used for progress and error reporting.
        tar_file_path: Path to the tar archive on disk.
        to_dataset: Dataset instance to populate.

    Returns:
        IsDataset | None: Populated dataset, or ``None`` when no serializer matches.
    """
    ...

register

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

Register a serializer class with the registry.

PARAMETER	DESCRIPTION
serializer_cls	Serializer class to add. TYPE: Type[IsSerializer]

Source code in src/omnipy/shared/protocols/data.py

def register(self, serializer_cls: Type[IsSerializer]) -> None:
    """Register a serializer class with the registry.

    Args:
        serializer_cls: Serializer class to add.
    """
    ...

IsSnapshotHolder

Bases: IsWeakKeyRefContainer[HasContentT, IsSnapshotWrapper[HasContentT, ContentT]], Protocol[HasContentT, ContentT]

              flowchart BT
              omnipy.shared.protocols.data.IsSnapshotHolder[IsSnapshotHolder]
              omnipy.shared.protocols._util.IsWeakKeyRefContainer[IsWeakKeyRefContainer]

                              omnipy.shared.protocols._util.IsWeakKeyRefContainer --> omnipy.shared.protocols.data.IsSnapshotHolder
                


              click omnipy.shared.protocols.data.IsSnapshotHolder href "" "omnipy.shared.protocols.data.IsSnapshotHolder"
              click omnipy.shared.protocols._util.IsWeakKeyRefContainer href "" "omnipy.shared.protocols._util.IsWeakKeyRefContainer"
            

Container protocol managing snapshots used for deepcopy/reactive tracking.

Returned by:

• Reference Code reference omnipy
  • data
    •  dataset  Dataset  snapshot_holder
    •  model  Model  snapshot_holder
    •  multi  MultiModelDataset  snapshot_holder
  • shared protocols  data  IsDataClassCreator  snapshot_holder

METHOD	DESCRIPTION
all_are_empty	Return whether every tracked snapshot collection is empty.
clear	Remove all tracked snapshots and bookkeeping state.
delete_scheduled_deepcopy_content_ids	Delete every content id previously scheduled for removal.
get
get_deepcopy_content_ids	Return tracked content ids participating in deepcopy bookkeeping.
get_deepcopy_content_ids_scheduled_for_deletion	Return tracked content ids waiting to be deleted.
schedule_deepcopy_content_ids_for_deletion	Mark tracked content ids for later deletion.
take_snapshot	Capture a snapshot of the given object's current content.
take_snapshot_setup	Prepare internal state before taking one or more snapshots.
take_snapshot_teardown	Finalize snapshot bookkeeping after snapshot capture completes.

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsSnapshotHolder(IsWeakKeyRefContainer[HasContentT, IsSnapshotWrapper[HasContentT, ContentT]],
                       Protocol[HasContentT, ContentT]):
    """Container protocol managing snapshots used for deepcopy/reactive tracking."""
    def clear(self) -> None:
        """Remove all tracked snapshots and bookkeeping state."""
        ...

    def all_are_empty(self, debug: bool = False) -> bool:
        """Return whether every tracked snapshot collection is empty.

        Args:
            debug: Whether extra diagnostics should be enabled during the check.

        Returns:
            bool: ``True`` when no tracked snapshot state remains.
        """
        ...

    def get_deepcopy_content_ids(self) -> SetDeque[int]:
        """Return tracked content ids participating in deepcopy bookkeeping.

        Returns:
            SetDeque[int]: Content ids currently tracked for deepcopy handling.
        """
        ...

    def get_deepcopy_content_ids_scheduled_for_deletion(self) -> SetDeque[int]:
        """Return tracked content ids waiting to be deleted.

        Returns:
            SetDeque[int]: Content ids marked for later deletion.
        """
        ...

    def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
        """Mark tracked content ids for later deletion.

        Args:
            keys: Content ids to queue for deletion.
        """
        ...

    def delete_scheduled_deepcopy_content_ids(self) -> None:
        """Delete every content id previously scheduled for removal."""
        ...

    def take_snapshot_setup(self) -> None:
        """Prepare internal state before taking one or more snapshots."""
        ...

    def take_snapshot_teardown(self) -> None:
        """Finalize snapshot bookkeeping after snapshot capture completes."""
        ...

    def take_snapshot(self, obj: HasContentT) -> None:
        """Capture a snapshot of the given object's current content.

        Args:
            obj: Object whose content should be snapshotted.
        """
        ...

all_are_empty

all_are_empty(debug: bool = False) -> bool

Return whether every tracked snapshot collection is empty.

PARAMETER	DESCRIPTION
debug	Whether extra diagnostics should be enabled during the check. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
bool	True when no tracked snapshot state remains. TYPE: bool

Source code in src/omnipy/shared/protocols/data.py

def all_are_empty(self, debug: bool = False) -> bool:
    """Return whether every tracked snapshot collection is empty.

    Args:
        debug: Whether extra diagnostics should be enabled during the check.

    Returns:
        bool: ``True`` when no tracked snapshot state remains.
    """
    ...

clear

clear() -> None

Remove all tracked snapshots and bookkeeping state.

Source code in src/omnipy/shared/protocols/data.py

def clear(self) -> None:
    """Remove all tracked snapshots and bookkeeping state."""
    ...

delete_scheduled_deepcopy_content_ids

delete_scheduled_deepcopy_content_ids() -> None

Delete every content id previously scheduled for removal.

Source code in src/omnipy/shared/protocols/data.py

def delete_scheduled_deepcopy_content_ids(self) -> None:
    """Delete every content id previously scheduled for removal."""
    ...

get

get(key: _AnyKeyT) -> _ValT | None

Source code in src/omnipy/shared/protocols/_util.py

def get(self, key: _AnyKeyT) -> _ValT | None:
    ...

get_deepcopy_content_ids

get_deepcopy_content_ids() -> SetDeque[int]

Return tracked content ids participating in deepcopy bookkeeping.

RETURNS	DESCRIPTION
SetDeque[int]	SetDeque[int]: Content ids currently tracked for deepcopy handling.

Source code in src/omnipy/shared/protocols/data.py

def get_deepcopy_content_ids(self) -> SetDeque[int]:
    """Return tracked content ids participating in deepcopy bookkeeping.

    Returns:
        SetDeque[int]: Content ids currently tracked for deepcopy handling.
    """
    ...

get_deepcopy_content_ids_scheduled_for_deletion

get_deepcopy_content_ids_scheduled_for_deletion() -> SetDeque[int]

Return tracked content ids waiting to be deleted.

RETURNS	DESCRIPTION
SetDeque[int]	SetDeque[int]: Content ids marked for later deletion.

Source code in src/omnipy/shared/protocols/data.py

def get_deepcopy_content_ids_scheduled_for_deletion(self) -> SetDeque[int]:
    """Return tracked content ids waiting to be deleted.

    Returns:
        SetDeque[int]: Content ids marked for later deletion.
    """
    ...

schedule_deepcopy_content_ids_for_deletion

schedule_deepcopy_content_ids_for_deletion(*keys: int) -> None

Mark tracked content ids for later deletion.

PARAMETER	DESCRIPTION
keys	Content ids to queue for deletion. TYPE: int DEFAULT: ()

Source code in src/omnipy/shared/protocols/data.py

def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
    """Mark tracked content ids for later deletion.

    Args:
        keys: Content ids to queue for deletion.
    """
    ...

take_snapshot

take_snapshot(obj: HasContentT) -> None

Capture a snapshot of the given object's current content.

PARAMETER	DESCRIPTION
obj	Object whose content should be snapshotted. TYPE: HasContentT

Source code in src/omnipy/shared/protocols/data.py

def take_snapshot(self, obj: HasContentT) -> None:
    """Capture a snapshot of the given object's current content.

    Args:
        obj: Object whose content should be snapshotted.
    """
    ...

take_snapshot_setup

take_snapshot_setup() -> None

Prepare internal state before taking one or more snapshots.

Source code in src/omnipy/shared/protocols/data.py

def take_snapshot_setup(self) -> None:
    """Prepare internal state before taking one or more snapshots."""
    ...

take_snapshot_teardown

take_snapshot_teardown() -> None

Finalize snapshot bookkeeping after snapshot capture completes.

Source code in src/omnipy/shared/protocols/data.py

def take_snapshot_teardown(self) -> None:
    """Finalize snapshot bookkeeping after snapshot capture completes."""
    ...

IsSnapshotWrapper

Bases: Protocol[ObjContraT, ContentT]

              flowchart BT
              omnipy.shared.protocols.data.IsSnapshotWrapper[IsSnapshotWrapper]

              

              click omnipy.shared.protocols.data.IsSnapshotWrapper href "" "omnipy.shared.protocols.data.IsSnapshotWrapper"
            

Snapshot record linking an object identity to captured content.

Subclassed by:

• Reference Code reference omnipy
  • data  snapshot  SnapshotHolder
  • shared protocols  data  IsSnapshotHolder

METHOD	DESCRIPTION
differs_from	Return whether the object's current content differs from the snapshot.
taken_of_same_obj	Return whether the snapshot belongs to the given object.

ATTRIBUTE	DESCRIPTION
id	TYPE: int
snapshot	TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsSnapshotWrapper(Protocol[ObjContraT, ContentT]):
    """Snapshot record linking an object identity to captured content."""

    id: int
    snapshot: ContentT

    def taken_of_same_obj(self, obj: ObjContraT) -> bool:
        """Return whether the snapshot belongs to the given object.

        Args:
            obj: Object to compare with the captured snapshot owner.

        Returns:
            bool: ``True`` when the snapshot was taken from ``obj``.
        """
        ...

    def differs_from(self, obj: ObjContraT) -> bool:
        """Return whether the object's current content differs from the snapshot.

        Args:
            obj: Object whose current content should be compared.

        Returns:
            bool: ``True`` when the current content no longer matches the snapshot.
        """
        ...

id instance-attribute

id: int

snapshot instance-attribute

snapshot: ContentT

differs_from

differs_from(obj: ObjContraT) -> bool

Return whether the object's current content differs from the snapshot.

PARAMETER	DESCRIPTION
obj	Object whose current content should be compared. TYPE: ObjContraT

RETURNS	DESCRIPTION
bool	True when the current content no longer matches the snapshot. TYPE: bool

Source code in src/omnipy/shared/protocols/data.py

def differs_from(self, obj: ObjContraT) -> bool:
    """Return whether the object's current content differs from the snapshot.

    Args:
        obj: Object whose current content should be compared.

    Returns:
        bool: ``True`` when the current content no longer matches the snapshot.
    """
    ...

taken_of_same_obj

taken_of_same_obj(obj: ObjContraT) -> bool

Return whether the snapshot belongs to the given object.

PARAMETER	DESCRIPTION
obj	Object to compare with the captured snapshot owner. TYPE: ObjContraT

RETURNS	DESCRIPTION
bool	True when the snapshot was taken from obj. TYPE: bool

Source code in src/omnipy/shared/protocols/data.py

def taken_of_same_obj(self, obj: ObjContraT) -> bool:
    """Return whether the snapshot belongs to the given object.

    Args:
        obj: Object to compare with the captured snapshot owner.

    Returns:
        bool: ``True`` when the snapshot was taken from ``obj``.
    """
    ...

IsTarFileSerializer

Bases: IsSerializer[_DatasetT], Protocol[_DatasetT]

              flowchart BT
              omnipy.shared.protocols.data.IsTarFileSerializer[IsTarFileSerializer]
              omnipy.shared.protocols.data.IsSerializer[IsSerializer]

                              omnipy.shared.protocols.data.IsSerializer --> omnipy.shared.protocols.data.IsTarFileSerializer
                


              click omnipy.shared.protocols.data.IsTarFileSerializer href "" "omnipy.shared.protocols.data.IsTarFileSerializer"
              click omnipy.shared.protocols.data.IsSerializer href "" "omnipy.shared.protocols.data.IsSerializer"
            

Serializer extension that stores dataset entries inside tar archives.

Returned by:

• Reference Code reference omnipy
  • data  serializer  SerializerRegistry
    •  tar_file_serializers
    •  detect_tar_file_serializers_from_dataset_cls
    •  detect_tar_file_serializers_from_file_suffix
  • shared protocols  data  IsSerializerRegistry
    •  tar_file_serializers
    •  detect_tar_file_serializers_from_dataset_cls
    •  detect_tar_file_serializers_from_file_suffix

METHOD	DESCRIPTION
create_dataset_from_tarfile	Populate a dataset from a tar archive payload.
create_tarfile_from_dataset	Create a tar archive payload from a dataset.
deserialize	Deserialize a bytes payload into a dataset instance.
get_dataset_cls_for_new	Return the dataset class this serializer creates when deserializing.
get_output_file_suffix	Return the default file suffix produced by this serializer.
is_dataset_directly_supported	Return whether the serializer can handle the dataset as-is.
serialize	Serialize a dataset into a bytes-like payload.

Source code in src/omnipy/shared/protocols/data.py

@runtime_checkable
class IsTarFileSerializer(IsSerializer[_DatasetT], Protocol[_DatasetT]):
    """Serializer extension that stores dataset entries inside tar archives."""
    @classmethod
    def create_tarfile_from_dataset(cls,
                                    dataset: _DatasetT,
                                    data_encode_func: Callable[..., bytes | memoryview]) -> bytes:
        """Create a tar archive payload from a dataset.

        Args:
            dataset: Dataset to archive.
            data_encode_func: Encoder used for individual dataset-entry payloads.

        Returns:
            bytes: Tar archive containing the serialized dataset entries.
        """

        ...

    @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 a dataset from a tar archive payload.

        Args:
            dataset: Dataset instance to populate.
            tarfile_bytes: Serialized tar archive payload.
            data_decode_func: Decoder used for individual archived payloads.
            dictify_object_func: Helper that converts decoded objects to importable values.
            import_method: Dataset import method to call for decoded entries.
            any_file_suffix: Whether suffix validation should be relaxed.
        """
        ...

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 a dataset from a tar archive payload.

PARAMETER	DESCRIPTION
dataset	Dataset instance to populate. TYPE: _DatasetT
tarfile_bytes	Serialized tar archive payload. TYPE: bytes
data_decode_func	Decoder used for individual archived payloads. TYPE: Callable[[IO[bytes]], Any]
dictify_object_func	Helper that converts decoded objects to importable values. TYPE: Callable[[str, Any], dict | str]
import_method	Dataset import method to call for decoded entries. TYPE: str DEFAULT: 'from_data'
any_file_suffix	Whether suffix validation should be relaxed. TYPE: bool DEFAULT: False

Source code in src/omnipy/shared/protocols/data.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 a dataset from a tar archive payload.

    Args:
        dataset: Dataset instance to populate.
        tarfile_bytes: Serialized tar archive payload.
        data_decode_func: Decoder used for individual archived payloads.
        dictify_object_func: Helper that converts decoded objects to importable values.
        import_method: Dataset import method to call for decoded entries.
        any_file_suffix: Whether suffix validation should be relaxed.
    """
    ...

create_tarfile_from_dataset classmethod

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

Create a tar archive payload from a dataset.

PARAMETER	DESCRIPTION
dataset	Dataset to archive. TYPE: _DatasetT
data_encode_func	Encoder used for individual dataset-entry payloads. TYPE: Callable[..., bytes | memoryview]

RETURNS	DESCRIPTION
bytes	Tar archive containing the serialized dataset entries. TYPE: bytes

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def create_tarfile_from_dataset(cls,
                                dataset: _DatasetT,
                                data_encode_func: Callable[..., bytes | memoryview]) -> bytes:
    """Create a tar archive payload from a dataset.

    Args:
        dataset: Dataset to archive.
        data_encode_func: Encoder used for individual dataset-entry payloads.

    Returns:
        bytes: Tar archive containing the serialized dataset entries.
    """

    ...

deserialize classmethod

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

Deserialize a bytes payload into a dataset instance.

PARAMETER	DESCRIPTION
serialized	Serialized dataset payload. TYPE: bytes
any_file_suffix	Whether suffix validation should be relaxed. DEFAULT: False

RETURNS	DESCRIPTION
_DatasetT	Deserialized dataset instance. TYPE: _DatasetT

Source code in src/omnipy/shared/protocols/data.py

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

    Args:
        serialized: Serialized dataset payload.
        any_file_suffix: Whether suffix validation should be relaxed.

    Returns:
        _DatasetT: Deserialized dataset instance.
    """
    ...

get_dataset_cls_for_new classmethod

get_dataset_cls_for_new() -> Type[IsDataset]

Return the dataset class this serializer creates when deserializing.

RETURNS	DESCRIPTION
Type[IsDataset]	Type[IsDataset]: Dataset class produced for fresh deserialization targets.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
    """Return the dataset class this serializer creates when deserializing.

    Returns:
        Type[IsDataset]: Dataset class produced for fresh deserialization targets.
    """
    ...

get_output_file_suffix classmethod

get_output_file_suffix() -> str

Return the default file suffix produced by this serializer.

RETURNS	DESCRIPTION
str	File suffix used for serialized output files. TYPE: str

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def get_output_file_suffix(cls) -> str:
    """Return the default file suffix produced by this serializer.

    Returns:
        str: File suffix used for serialized output files.
    """
    ...

is_dataset_directly_supported classmethod

is_dataset_directly_supported(dataset: IsDataset) -> bool

Return whether the serializer can handle the dataset as-is.

PARAMETER	DESCRIPTION
dataset	Dataset instance to check. TYPE: IsDataset

RETURNS	DESCRIPTION
bool	True when no dataset conversion is required before serialization. TYPE: bool

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    """Return whether the serializer can handle the dataset as-is.

    Args:
        dataset: Dataset instance to check.

    Returns:
        bool: ``True`` when no dataset conversion is required before serialization.
    """
    ...

serialize classmethod

serialize(dataset: _DatasetT) -> bytes | memoryview

Serialize a dataset into a bytes-like payload.

PARAMETER	DESCRIPTION
dataset	Dataset instance to serialize. TYPE: _DatasetT

RETURNS	DESCRIPTION
bytes | memoryview	bytes | memoryview: Serialized dataset payload.

Source code in src/omnipy/shared/protocols/data.py

@classmethod
def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
    """Serialize a dataset into a bytes-like payload.

    Args:
        dataset: Dataset instance to serialize.

    Returns:
        bytes | memoryview: Serialized dataset payload.
    """
    ...