Skip to content

omnipy.shared.protocols.data

CLASS DESCRIPTION
AvailableDisplayDims
HasContent
HasData
IsDataClassCreator
IsDataset

Dict-based container of data files that follow a specific Model
IsFailedData
IsHttpUrlDataset
IsHttpUrlModel
IsModel
IsMultiModelDataset

Variant of Dataset that allows custom models to be set on individual data files
IsPendingData
IsReactive
IsReactiveObjects
IsSerializer
IsSerializerRegistry
IsSnapshotHolder
IsSnapshotWrapper
IsTarFileSerializer
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)

HasContentT module-attribute 

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

IsPathOrUrl module-attribute 

IsPathOrUrl: TypeAlias = 'str | IsHttpUrlModel'

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'

ObjContraT module-attribute 

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

AvailableDisplayDims

Bases: TypedDict

ATTRIBUTE DESCRIPTION
height

TYPE: pyd.NonNegativeInt | None

width

TYPE: pyd.NonNegativeInt | None

Source code in src/omnipy/shared/protocols/data.py 
364
365
366
class AvailableDisplayDims(TypedDict):
    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]

ATTRIBUTE DESCRIPTION
content

TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py 
65
66
67
68
69
70
71
72
73
@runtime_checkable
class HasContent(Protocol[ContentT]):
    @property
    def content(self) -> ContentT:
        ...

    @content.setter
    def content(self, value: ContentT) -> None:
        ...

content property writable 

content: ContentT

HasData

Bases: Protocol

ATTRIBUTE DESCRIPTION
data

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

Source code in src/omnipy/shared/protocols/data.py 
60
61
62
@runtime_checkable
class HasData(Protocol):
    data: dict[str, Any | IsPendingData | IsFailedData]

data instance-attribute 

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

IsDataClassCreator

Bases: Protocol[HasContentT, ContentT]

METHOD DESCRIPTION
deepcopy_context
set_config
set_reactive_objects
ATTRIBUTE DESCRIPTION
config

TYPE: IsDataConfig

reactive_objects

TYPE: IsReactiveObjects | None

snapshot_holder

TYPE: IsSnapshotHolder[HasContentT, ContentT]

Source code in src/omnipy/shared/protocols/data.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
@runtime_checkable
class IsDataClassCreator(Protocol[HasContentT, ContentT]):
    """"""
    @property
    def config(self) -> IsDataConfig:
        ...

    def set_config(self, config: IsDataConfig) -> None:
        ...

    @property
    def reactive_objects(self) -> IsReactiveObjects | None:
        ...

    def set_reactive_objects(self, reactive_objects: IsReactiveObjects) -> None:
        ...

    @property
    def snapshot_holder(self) -> IsSnapshotHolder[HasContentT, ContentT]:
        ...

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

config property 

config: IsDataConfig

reactive_objects property 

reactive_objects: IsReactiveObjects | None

snapshot_holder property 

snapshot_holder: IsSnapshotHolder[HasContentT, ContentT]

deepcopy_context 

deepcopy_context(
    top_level_entry_func: Callable[[], None], top_level_exit_func: Callable[[], None]
) -> ContextManager[int]
Source code in src/omnipy/shared/protocols/data.py 
411
412
413
414
415
416
def deepcopy_context(
    self,
    top_level_entry_func: Callable[[], None],
    top_level_exit_func: Callable[[], None],
) -> ContextManager[int]:
    ...

set_config 

set_config(config: IsDataConfig) -> None
Source code in src/omnipy/shared/protocols/data.py 
397
398
def set_config(self, config: IsDataConfig) -> None:
    ...

set_reactive_objects 

set_reactive_objects(reactive_objects: IsReactiveObjects) -> None
Source code in src/omnipy/shared/protocols/data.py 
404
405
def set_reactive_objects(self, reactive_objects: IsReactiveObjects) -> None:
    ...

IsDataset

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

Dict-based container of data files that follow a specific Model

METHOD DESCRIPTION
__init__
clear

D.clear() -> None. Remove all items from D.
failed_task_details
from_data
from_json
get

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

Returns the concrete type (Model or Dataset class) used for all
keys

D.keys() -> a set-like object providing a view on D's keys
load
load_into
pending_task_details
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
setdefault

D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
to_data
to_json
to_json_schema
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

TYPE: Self

failed_data

TYPE: Self

pending_data

TYPE: Self

Source code in src/omnipy/shared/protocols/data.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@runtime_checkable
class IsDataset(IsMutableMapping[str, _ModelOrDatasetT], Protocol[_ModelOrDatasetT]):
    """
    Dict-based container of data files that follow a specific Model
    """
    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]:
        """
        Returns the concrete type (Model or Dataset class) used for all
        data files in the dataset, e.g.: `Model[list[int]]`, or
        `Dataset[Model[dict[str, float]]]` for nested datasets.
        :return: The concrete type (Model or Dataset class) used for all
                 data files in the dataset.
        """
        ...

    def to_data(self) -> dict[str, Any]:
        ...

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

    def to_json(self, pretty=True) -> dict[str, str]:
        ...

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

    @classmethod
    def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
        ...

    def save(self, path: str) -> None:
        ...

    @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]:
        ...

    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]:
        ...

    @property
    def available_data(self) -> Self:
        ...

    @property
    def pending_data(self) -> Self:
        ...

    @property
    def failed_data(self) -> Self:
        ...

    def pending_task_details(self) -> dict[str, IsPendingData]:
        ...

    def failed_task_details(self) -> dict[str, IsFailedData]:
        ...

    # 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

failed_data property 

failed_data: Self

pending_data property 

pending_data: 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 
88
89
90
91
92
93
94
95
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 
990
991
992
993
994
def clear(self) -> None:
    """
    D.clear() -> None.  Remove all items from D.
    """
    raise AssumedToBeImplementedException

failed_task_details 

failed_task_details() -> dict[str, IsFailedData]
Source code in src/omnipy/shared/protocols/data.py 
166
167
def failed_task_details(self) -> dict[str, IsFailedData]:
    ...

from_data 

from_data(data: Mapping[str, Any] | Iterable[tuple[str, Any]], update: bool = True) -> None
Source code in src/omnipy/shared/protocols/data.py 
112
113
114
115
def from_data(self,
              data: Mapping[str, Any] | Iterable[tuple[str, Any]],
              update: bool = True) -> None:
    ...

from_json 

from_json(data: Mapping[str, str] | Iterable[tuple[str, str]], update: bool = True) -> None
Source code in src/omnipy/shared/protocols/data.py 
120
121
122
123
def from_json(self,
              data: Mapping[str, str] | Iterable[tuple[str, str]],
              update: bool = True) -> None:
    ...

get 

get(key: _KT) -> _VT_co | None

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

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

get(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.

Source code in src/omnipy/shared/protocols/typing.py 
949
950
951
952
953
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]

Returns the concrete type (Model or Dataset class) used for all data files in the dataset, e.g.: Model[list[int]], or Dataset[Model[dict[str, float]]] for nested datasets. :return: The concrete type (Model or Dataset class) used for all data files in the dataset.

Source code in src/omnipy/shared/protocols/data.py 
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
@functools.cache
def get_type(cls) -> type[_ModelOrDatasetT]:
    """
    Returns the concrete type (Model or Dataset class) used for all
    data files in the dataset, e.g.: `Model[list[int]]`, or
    `Dataset[Model[dict[str, float]]]` for nested datasets.
    :return: The concrete type (Model or Dataset class) used for all
             data files in the dataset.
    """
    ...

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 
962
963
964
965
966
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]
Source code in src/omnipy/shared/protocols/data.py 
132
133
134
135
136
137
138
139
140
@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_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]
Source code in src/omnipy/shared/protocols/data.py 
142
143
144
145
146
147
148
149
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]:
    ...

pending_task_details 

pending_task_details() -> dict[str, IsPendingData]
Source code in src/omnipy/shared/protocols/data.py 
163
164
def pending_task_details(self) -> dict[str, IsPendingData]:
    ...

pop 

pop(key: _KT) -> _VT

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

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

pop(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.

Source code in src/omnipy/shared/protocols/typing.py 
1002
1003
1004
1005
1006
1007
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 
1009
1010
1011
1012
1013
1014
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
Source code in src/omnipy/shared/protocols/data.py 
129
130
def save(self, path: str) -> None:
    ...

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 
1025
1026
1027
1028
1029
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]
Source code in src/omnipy/shared/protocols/data.py 
109
110
def to_data(self) -> dict[str, Any]:
    ...

to_json 

to_json(pretty=True) -> dict[str, str]
Source code in src/omnipy/shared/protocols/data.py 
117
118
def to_json(self, pretty=True) -> dict[str, str]:
    ...

to_json_schema classmethod 

to_json_schema(pretty=True) -> str | dict[str, str]
Source code in src/omnipy/shared/protocols/data.py 
125
126
127
@classmethod
def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
    ...

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

update(
    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

Source code in src/omnipy/shared/protocols/typing.py 
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
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 
968
969
970
971
972
def values(self) -> IsValuesView[_VT_co]:
    """
    D.values() -> an object providing a view on D's values
    """
    raise AssumedToBeImplementedException

IsFailedData dataclass

Bases: Protocol

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 
52
53
54
55
56
57
@runtime_checkable
@dataclass(frozen=True, kw_only=True)
class IsFailedData(Protocol):
    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

METHOD DESCRIPTION
__init__
clear

D.clear() -> None. Remove all items from D.
failed_task_details
from_data
from_json
get

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

Returns the concrete type (Model or Dataset class) used for all
keys

D.keys() -> a set-like object providing a view on D's keys
load
load_into
pending_task_details
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
setdefault

D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
to_data
to_json
to_json_schema
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

TYPE: Self

failed_data

TYPE: Self

pending_data

TYPE: Self

Source code in src/omnipy/shared/protocols/data.py 
222
223
224
@runtime_checkable
class IsHttpUrlDataset(IsDataset, Protocol):
    ...

available_data property 

available_data: Self

failed_data property 

failed_data: Self

pending_data property 

pending_data: 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 
88
89
90
91
92
93
94
95
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 
990
991
992
993
994
def clear(self) -> None:
    """
    D.clear() -> None.  Remove all items from D.
    """
    raise AssumedToBeImplementedException

failed_task_details 

failed_task_details() -> dict[str, IsFailedData]
Source code in src/omnipy/shared/protocols/data.py 
166
167
def failed_task_details(self) -> dict[str, IsFailedData]:
    ...

from_data 

from_data(data: Mapping[str, Any] | Iterable[tuple[str, Any]], update: bool = True) -> None
Source code in src/omnipy/shared/protocols/data.py 
112
113
114
115
def from_data(self,
              data: Mapping[str, Any] | Iterable[tuple[str, Any]],
              update: bool = True) -> None:
    ...

from_json 

from_json(data: Mapping[str, str] | Iterable[tuple[str, str]], update: bool = True) -> None
Source code in src/omnipy/shared/protocols/data.py 
120
121
122
123
def from_json(self,
              data: Mapping[str, str] | Iterable[tuple[str, str]],
              update: bool = True) -> None:
    ...

get 

get(key: _KT) -> _VT_co | None

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

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

get(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.

Source code in src/omnipy/shared/protocols/typing.py 
949
950
951
952
953
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]

Returns the concrete type (Model or Dataset class) used for all data files in the dataset, e.g.: Model[list[int]], or Dataset[Model[dict[str, float]]] for nested datasets. :return: The concrete type (Model or Dataset class) used for all data files in the dataset.

Source code in src/omnipy/shared/protocols/data.py 
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
@functools.cache
def get_type(cls) -> type[_ModelOrDatasetT]:
    """
    Returns the concrete type (Model or Dataset class) used for all
    data files in the dataset, e.g.: `Model[list[int]]`, or
    `Dataset[Model[dict[str, float]]]` for nested datasets.
    :return: The concrete type (Model or Dataset class) used for all
             data files in the dataset.
    """
    ...

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 
962
963
964
965
966
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]
Source code in src/omnipy/shared/protocols/data.py 
132
133
134
135
136
137
138
139
140
@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_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]
Source code in src/omnipy/shared/protocols/data.py 
142
143
144
145
146
147
148
149
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]:
    ...

pending_task_details 

pending_task_details() -> dict[str, IsPendingData]
Source code in src/omnipy/shared/protocols/data.py 
163
164
def pending_task_details(self) -> dict[str, IsPendingData]:
    ...

pop 

pop(key: _KT) -> _VT

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

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

pop(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.

Source code in src/omnipy/shared/protocols/typing.py 
1002
1003
1004
1005
1006
1007
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 
1009
1010
1011
1012
1013
1014
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
Source code in src/omnipy/shared/protocols/data.py 
129
130
def save(self, path: str) -> None:
    ...

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 
1025
1026
1027
1028
1029
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]
Source code in src/omnipy/shared/protocols/data.py 
109
110
def to_data(self) -> dict[str, Any]:
    ...

to_json 

to_json(pretty=True) -> dict[str, str]
Source code in src/omnipy/shared/protocols/data.py 
117
118
def to_json(self, pretty=True) -> dict[str, str]:
    ...

to_json_schema classmethod 

to_json_schema(pretty=True) -> str | dict[str, str]
Source code in src/omnipy/shared/protocols/data.py 
125
126
127
@classmethod
def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
    ...

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

update(
    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

Source code in src/omnipy/shared/protocols/typing.py 
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
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 
968
969
970
971
972
def values(self) -> IsValuesView[_VT_co]:
    """
    D.values() -> an object providing a view on D's values
    """
    raise AssumedToBeImplementedException

IsHttpUrlModel

Bases: IsModel, Protocol

METHOD DESCRIPTION
full_type
ATTRIBUTE DESCRIPTION
content

TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py 
217
218
219
@runtime_checkable
class IsHttpUrlModel(IsModel, Protocol):
    ...

content property writable 

content: ContentT

full_type classmethod 

full_type() -> type[_RootT]
Source code in src/omnipy/shared/protocols/data.py 
78
79
80
@classmethod
def full_type(cls) -> type[_RootT]:
    ...

IsModel

Bases: HasContent[_RootT], Protocol[_RootT]

METHOD DESCRIPTION
full_type
ATTRIBUTE DESCRIPTION
content

TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py 
76
77
78
79
80
@runtime_checkable
class IsModel(HasContent[_RootT], Protocol[_RootT]):
    @classmethod
    def full_type(cls) -> type[_RootT]:
        ...

content property writable 

content: ContentT

full_type classmethod 

full_type() -> type[_RootT]
Source code in src/omnipy/shared/protocols/data.py 
78
79
80
@classmethod
def full_type(cls) -> type[_RootT]:
    ...

IsMultiModelDataset

Bases: IsDataset[_ModelOrDatasetT], Protocol[_ModelOrDatasetT]

Variant of Dataset that allows custom models to be set on individual data files

METHOD DESCRIPTION
__init__
clear

D.clear() -> None. Remove all items from D.
failed_task_details
from_data
from_json
get

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

Returns the concrete type (Model or Dataset class) used for all
keys

D.keys() -> a set-like object providing a view on D's keys
load
load_into
pending_task_details
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
set_model
setdefault

D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D
to_data
to_json
to_json_schema
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

TYPE: Self

failed_data

TYPE: Self

pending_data

TYPE: Self

Source code in src/omnipy/shared/protocols/data.py 
205
206
207
208
209
210
211
212
213
214
@runtime_checkable
class IsMultiModelDataset(IsDataset[_ModelOrDatasetT], Protocol[_ModelOrDatasetT]):
    """
        Variant of Dataset that allows custom models to be set on individual data files
    """
    def set_model(self, data_file: str, model: type[IsModel]) -> None:
        ...

    def get_model(self, data_file: str) -> type[IsModel]:
        ...

available_data property 

available_data: Self

failed_data property 

failed_data: Self

pending_data property 

pending_data: 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 
88
89
90
91
92
93
94
95
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 
990
991
992
993
994
def clear(self) -> None:
    """
    D.clear() -> None.  Remove all items from D.
    """
    raise AssumedToBeImplementedException

failed_task_details 

failed_task_details() -> dict[str, IsFailedData]
Source code in src/omnipy/shared/protocols/data.py 
166
167
def failed_task_details(self) -> dict[str, IsFailedData]:
    ...

from_data 

from_data(data: Mapping[str, Any] | Iterable[tuple[str, Any]], update: bool = True) -> None
Source code in src/omnipy/shared/protocols/data.py 
112
113
114
115
def from_data(self,
              data: Mapping[str, Any] | Iterable[tuple[str, Any]],
              update: bool = True) -> None:
    ...

from_json 

from_json(data: Mapping[str, str] | Iterable[tuple[str, str]], update: bool = True) -> None
Source code in src/omnipy/shared/protocols/data.py 
120
121
122
123
def from_json(self,
              data: Mapping[str, str] | Iterable[tuple[str, str]],
              update: bool = True) -> None:
    ...

get 

get(key: _KT) -> _VT_co | None

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

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

get(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.

Source code in src/omnipy/shared/protocols/typing.py 
949
950
951
952
953
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]
Source code in src/omnipy/shared/protocols/data.py 
213
214
def get_model(self, data_file: str) -> type[IsModel]:
    ...

get_type cached classmethod 

get_type() -> type[_ModelOrDatasetT]

Returns the concrete type (Model or Dataset class) used for all data files in the dataset, e.g.: Model[list[int]], or Dataset[Model[dict[str, float]]] for nested datasets. :return: The concrete type (Model or Dataset class) used for all data files in the dataset.

Source code in src/omnipy/shared/protocols/data.py 
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
@functools.cache
def get_type(cls) -> type[_ModelOrDatasetT]:
    """
    Returns the concrete type (Model or Dataset class) used for all
    data files in the dataset, e.g.: `Model[list[int]]`, or
    `Dataset[Model[dict[str, float]]]` for nested datasets.
    :return: The concrete type (Model or Dataset class) used for all
             data files in the dataset.
    """
    ...

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 
962
963
964
965
966
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]
Source code in src/omnipy/shared/protocols/data.py 
132
133
134
135
136
137
138
139
140
@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_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]
Source code in src/omnipy/shared/protocols/data.py 
142
143
144
145
146
147
148
149
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]:
    ...

pending_task_details 

pending_task_details() -> dict[str, IsPendingData]
Source code in src/omnipy/shared/protocols/data.py 
163
164
def pending_task_details(self) -> dict[str, IsPendingData]:
    ...

pop 

pop(key: _KT) -> _VT

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

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

pop(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.

Source code in src/omnipy/shared/protocols/typing.py 
1002
1003
1004
1005
1006
1007
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 
1009
1010
1011
1012
1013
1014
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
Source code in src/omnipy/shared/protocols/data.py 
129
130
def save(self, path: str) -> None:
    ...

set_model 

set_model(data_file: str, model: type[IsModel]) -> None
Source code in src/omnipy/shared/protocols/data.py 
210
211
def set_model(self, data_file: str, model: type[IsModel]) -> None:
    ...

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 
1025
1026
1027
1028
1029
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]
Source code in src/omnipy/shared/protocols/data.py 
109
110
def to_data(self) -> dict[str, Any]:
    ...

to_json 

to_json(pretty=True) -> dict[str, str]
Source code in src/omnipy/shared/protocols/data.py 
117
118
def to_json(self, pretty=True) -> dict[str, str]:
    ...

to_json_schema classmethod 

to_json_schema(pretty=True) -> str | dict[str, str]
Source code in src/omnipy/shared/protocols/data.py 
125
126
127
@classmethod
def to_json_schema(cls, pretty=True) -> str | dict[str, str]:
    ...

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

update(
    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

Source code in src/omnipy/shared/protocols/typing.py 
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
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 
968
969
970
971
972
def values(self) -> IsValuesView[_VT_co]:
    """
    D.values() -> an object providing a view on D's values
    """
    raise AssumedToBeImplementedException

IsPendingData dataclass

Bases: Protocol

METHOD DESCRIPTION
__init__
ATTRIBUTE DESCRIPTION
job_name

TYPE: str

job_unique_name

TYPE: str

Source code in src/omnipy/shared/protocols/data.py 
45
46
47
48
49
@runtime_checkable
@dataclass(frozen=True, kw_only=True)
class IsPendingData(Protocol):
    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]

METHOD DESCRIPTION
set
ATTRIBUTE DESCRIPTION
value

TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py 
369
370
371
372
373
374
375
376
@runtime_checkable
class IsReactive(Protocol[ContentT]):
    @property
    def value(self) -> ContentT:
        ...

    def set(self, value: ContentT):
        ...

value property 

value: ContentT

set 

set(value: ContentT)
Source code in src/omnipy/shared/protocols/data.py 
375
376
def set(self, value: ContentT):
    ...

IsReactiveObjects

Bases: Protocol

ATTRIBUTE DESCRIPTION
available_display_dims_in_px

TYPE: IsReactive[AvailableDisplayDims]

jupyter_ui_config

TYPE: IsReactive[IsJupyterUserInterfaceConfig]

layout_config

TYPE: IsReactive[IsLayoutConfig]

text_config

TYPE: IsReactive[IsTextConfig]

Source code in src/omnipy/shared/protocols/data.py 
379
380
381
382
383
384
385
386
387
@runtime_checkable
class IsReactiveObjects(Protocol):
    jupyter_ui_config: IsReactive[IsJupyterUserInterfaceConfig]
    text_config: IsReactive[IsTextConfig]
    layout_config: IsReactive[IsLayoutConfig]
    available_display_dims_in_px: IsReactive[AvailableDisplayDims]

    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]

text_config instance-attribute 

text_config: IsReactive[IsTextConfig]

IsSerializer

Bases: Protocol[_DatasetT]

METHOD DESCRIPTION
deserialize
get_dataset_cls_for_new
get_output_file_suffix
is_dataset_directly_supported
serialize
Source code in src/omnipy/shared/protocols/data.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
class IsSerializer(Protocol[_DatasetT]):
    """"""
    @classmethod
    def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
        ...

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

    @classmethod
    def get_output_file_suffix(cls) -> str:
        ...

    @classmethod
    def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
        ...

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

deserialize classmethod 

deserialize(serialized: bytes, any_file_suffix=False) -> _DatasetT
Source code in src/omnipy/shared/protocols/data.py 
245
246
247
@classmethod
def deserialize(cls, serialized: bytes, any_file_suffix=False) -> _DatasetT:
    ...

get_dataset_cls_for_new classmethod 

get_dataset_cls_for_new() -> Type[IsDataset]
Source code in src/omnipy/shared/protocols/data.py 
233
234
235
@classmethod
def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
    ...

get_output_file_suffix classmethod 

get_output_file_suffix() -> str
Source code in src/omnipy/shared/protocols/data.py 
237
238
239
@classmethod
def get_output_file_suffix(cls) -> str:
    ...

is_dataset_directly_supported classmethod 

is_dataset_directly_supported(dataset: IsDataset) -> bool
Source code in src/omnipy/shared/protocols/data.py 
229
230
231
@classmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    ...

serialize classmethod 

serialize(dataset: _DatasetT) -> bytes | memoryview
Source code in src/omnipy/shared/protocols/data.py 
241
242
243
@classmethod
def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
    ...

IsSerializerRegistry

Bases: Protocol

METHOD DESCRIPTION
__init__
auto_detect
auto_detect_tar_file_serializer
detect_tar_file_serializers_from_dataset_cls
detect_tar_file_serializers_from_file_suffix
load_from_tar_file_path_based_on_dataset_cls
load_from_tar_file_path_based_on_file_suffix
register
ATTRIBUTE DESCRIPTION
serializers

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

tar_file_serializers

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

Source code in src/omnipy/shared/protocols/data.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
@runtime_checkable
class IsSerializerRegistry(Protocol):
    """"""
    def __init__(self) -> None:
        ...

    def register(self, serializer_cls: Type[IsSerializer]) -> None:
        ...

    @property
    def serializers(self) -> tuple[Type[IsSerializer], ...]:
        ...

    @property
    def tar_file_serializers(self) -> tuple[Type[IsTarFileSerializer], ...]:
        ...

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

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

    @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], ...]:
        ...

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

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

    def load_from_tar_file_path_based_on_dataset_cls(self,
                                                     log_obj: CanLog,
                                                     tar_file_path: str,
                                                     to_dataset: IsDataset) -> IsDataset | None:
        ...

serializers property 

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

tar_file_serializers property 

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

__init__ 

__init__() -> None
Source code in src/omnipy/shared/protocols/data.py 
273
274
def __init__(self) -> None:
    ...

auto_detect 

auto_detect(dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]
Source code in src/omnipy/shared/protocols/data.py 
287
288
def auto_detect(self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
    ...

auto_detect_tar_file_serializer 

auto_detect_tar_file_serializer(
    dataset: IsDataset,
) -> tuple[IsDataset, IsSerializer] | tuple[None, None]
Source code in src/omnipy/shared/protocols/data.py 
290
291
292
def auto_detect_tar_file_serializer(
        self, dataset: IsDataset) -> tuple[IsDataset, IsSerializer] | tuple[None, None]:
    ...

detect_tar_file_serializers_from_dataset_cls 

detect_tar_file_serializers_from_dataset_cls(
    dataset: IsDataset,
) -> tuple[Type[IsTarFileSerializer], ...]
Source code in src/omnipy/shared/protocols/data.py 
299
300
301
def detect_tar_file_serializers_from_dataset_cls(
        self, dataset: IsDataset) -> tuple[Type[IsTarFileSerializer], ...]:
    ...

detect_tar_file_serializers_from_file_suffix 

detect_tar_file_serializers_from_file_suffix(
    file_suffix: str,
) -> tuple[Type[IsTarFileSerializer], ...]
Source code in src/omnipy/shared/protocols/data.py 
303
304
305
def detect_tar_file_serializers_from_file_suffix(
        self, file_suffix: str) -> tuple[Type[IsTarFileSerializer], ...]:
    ...

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
Source code in src/omnipy/shared/protocols/data.py 
313
314
315
316
317
def load_from_tar_file_path_based_on_dataset_cls(self,
                                                 log_obj: CanLog,
                                                 tar_file_path: str,
                                                 to_dataset: IsDataset) -> IsDataset | None:
    ...

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
Source code in src/omnipy/shared/protocols/data.py 
307
308
309
310
311
def load_from_tar_file_path_based_on_file_suffix(self,
                                                 log_obj: CanLog,
                                                 tar_file_path: str,
                                                 to_dataset: IsDataset) -> IsDataset | None:
    ...

register 

register(serializer_cls: Type[IsSerializer]) -> None
Source code in src/omnipy/shared/protocols/data.py 
276
277
def register(self, serializer_cls: Type[IsSerializer]) -> None:
    ...

IsSnapshotHolder

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

METHOD DESCRIPTION
all_are_empty
clear
delete_scheduled_deepcopy_content_ids
get
get_deepcopy_content_ids
get_deepcopy_content_ids_scheduled_for_deletion
schedule_deepcopy_content_ids_for_deletion
take_snapshot
take_snapshot_setup
take_snapshot_teardown
Source code in src/omnipy/shared/protocols/data.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
@runtime_checkable
class IsSnapshotHolder(IsWeakKeyRefContainer[HasContentT, IsSnapshotWrapper[HasContentT, ContentT]],
                       Protocol[HasContentT, ContentT]):
    """"""
    def clear(self) -> None:
        ...

    def all_are_empty(self, debug: bool = False) -> bool:
        ...

    def get_deepcopy_content_ids(self) -> SetDeque[int]:
        ...

    def get_deepcopy_content_ids_scheduled_for_deletion(self) -> SetDeque[int]:
        ...

    def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
        ...

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

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

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

    def take_snapshot(self, obj: HasContentT) -> None:
        ...

all_are_empty 

all_are_empty(debug: bool = False) -> bool
Source code in src/omnipy/shared/protocols/data.py 
339
340
def all_are_empty(self, debug: bool = False) -> bool:
    ...

clear 

clear() -> None
Source code in src/omnipy/shared/protocols/data.py 
336
337
def clear(self) -> None:
    ...

delete_scheduled_deepcopy_content_ids 

delete_scheduled_deepcopy_content_ids() -> None
Source code in src/omnipy/shared/protocols/data.py 
351
352
def delete_scheduled_deepcopy_content_ids(self) -> None:
    ...

get 

get(key: _AnyKeyT) -> _ValT | None
Source code in src/omnipy/shared/protocols/_util.py 
20
21
def get(self, key: _AnyKeyT) -> _ValT | None:
    ...

get_deepcopy_content_ids 

get_deepcopy_content_ids() -> SetDeque[int]
Source code in src/omnipy/shared/protocols/data.py 
342
343
def get_deepcopy_content_ids(self) -> SetDeque[int]:
    ...

get_deepcopy_content_ids_scheduled_for_deletion 

get_deepcopy_content_ids_scheduled_for_deletion() -> SetDeque[int]
Source code in src/omnipy/shared/protocols/data.py 
345
346
def get_deepcopy_content_ids_scheduled_for_deletion(self) -> SetDeque[int]:
    ...

schedule_deepcopy_content_ids_for_deletion 

schedule_deepcopy_content_ids_for_deletion(*keys: int) -> None
Source code in src/omnipy/shared/protocols/data.py 
348
349
def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
    ...

take_snapshot 

take_snapshot(obj: HasContentT) -> None
Source code in src/omnipy/shared/protocols/data.py 
360
361
def take_snapshot(self, obj: HasContentT) -> None:
    ...

take_snapshot_setup 

take_snapshot_setup() -> None
Source code in src/omnipy/shared/protocols/data.py 
354
355
def take_snapshot_setup(self) -> None:
    ...

take_snapshot_teardown 

take_snapshot_teardown() -> None
Source code in src/omnipy/shared/protocols/data.py 
357
358
def take_snapshot_teardown(self) -> None:
    ...

IsSnapshotWrapper

Bases: Protocol[ObjContraT, ContentT]

METHOD DESCRIPTION
differs_from
taken_of_same_obj
ATTRIBUTE DESCRIPTION
id

TYPE: int

snapshot

TYPE: ContentT

Source code in src/omnipy/shared/protocols/data.py 
320
321
322
323
324
325
326
327
328
329
@runtime_checkable
class IsSnapshotWrapper(Protocol[ObjContraT, ContentT]):
    id: int
    snapshot: ContentT

    def taken_of_same_obj(self, obj: ObjContraT) -> bool:
        ...

    def differs_from(self, obj: ObjContraT) -> bool:
        ...

id instance-attribute 

id: int

snapshot instance-attribute 

snapshot: ContentT

differs_from 

differs_from(obj: ObjContraT) -> bool
Source code in src/omnipy/shared/protocols/data.py 
328
329
def differs_from(self, obj: ObjContraT) -> bool:
    ...

taken_of_same_obj 

taken_of_same_obj(obj: ObjContraT) -> bool
Source code in src/omnipy/shared/protocols/data.py 
325
326
def taken_of_same_obj(self, obj: ObjContraT) -> bool:
    ...

IsTarFileSerializer

Bases: IsSerializer[_DatasetT], Protocol[_DatasetT]

METHOD DESCRIPTION
create_dataset_from_tarfile
create_tarfile_from_dataset
deserialize
get_dataset_cls_for_new
get_output_file_suffix
is_dataset_directly_supported
serialize
Source code in src/omnipy/shared/protocols/data.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
@runtime_checkable
class IsTarFileSerializer(IsSerializer[_DatasetT], Protocol[_DatasetT]):
    @classmethod
    def create_tarfile_from_dataset(cls,
                                    dataset: _DatasetT,
                                    data_encode_func: Callable[..., bytes | memoryview]) -> bytes:
        """"""
        ...

    @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:
        ...

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
Source code in src/omnipy/shared/protocols/data.py 
259
260
261
262
263
264
265
266
267
@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:
    ...

create_tarfile_from_dataset classmethod 

create_tarfile_from_dataset(
    dataset: _DatasetT, data_encode_func: Callable[..., bytes | memoryview]
) -> bytes
Source code in src/omnipy/shared/protocols/data.py 
252
253
254
255
256
257
@classmethod
def create_tarfile_from_dataset(cls,
                                dataset: _DatasetT,
                                data_encode_func: Callable[..., bytes | memoryview]) -> bytes:
    """"""
    ...

deserialize classmethod 

deserialize(serialized: bytes, any_file_suffix=False) -> _DatasetT
Source code in src/omnipy/shared/protocols/data.py 
245
246
247
@classmethod
def deserialize(cls, serialized: bytes, any_file_suffix=False) -> _DatasetT:
    ...

get_dataset_cls_for_new classmethod 

get_dataset_cls_for_new() -> Type[IsDataset]
Source code in src/omnipy/shared/protocols/data.py 
233
234
235
@classmethod
def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
    ...

get_output_file_suffix classmethod 

get_output_file_suffix() -> str
Source code in src/omnipy/shared/protocols/data.py 
237
238
239
@classmethod
def get_output_file_suffix(cls) -> str:
    ...

is_dataset_directly_supported classmethod 

is_dataset_directly_supported(dataset: IsDataset) -> bool
Source code in src/omnipy/shared/protocols/data.py 
229
230
231
@classmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    ...

serialize classmethod 

serialize(dataset: _DatasetT) -> bytes | memoryview
Source code in src/omnipy/shared/protocols/data.py 
241
242
243
@classmethod
def serialize(cls, dataset: _DatasetT) -> bytes | memoryview:
    ...