Skip to content

omnipy.data.helpers

CLASS DESCRIPTION
DoubleTypeVarStore
FailedData
MethodInfo
PendingData
ResetSolutionTuple
TypeVarStore
TypeVarStore1
TypeVarStore2
TypeVarStore3
TypeVarStore4
YesNoMaybe
FUNCTION DESCRIPTION
build_own_module_and_global_namespace_for_forward_refs

Build own-module and global namespaces for fwd reference resolution
cleanup_name_qualname_and_module
ATTRIBUTE DESCRIPTION
SPECIAL_METHODS_INFO_DICT

TYPE: dict[str, MethodInfo]

SPECIAL_METHODS_INFO_DICT module-attribute 

SPECIAL_METHODS_INFO_DICT: dict[str, MethodInfo] = {
    "__len__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__length_hint__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__getitem__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__setitem__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__delitem__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__missing__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__iter__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__next__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__reversed__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__contains__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__add__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.YES),
    "__sub__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__mul__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__matmul__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__truediv__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__floordiv__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__mod__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__divmod__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__pow__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__lshift__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rshift__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__and__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__xor__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__or__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__radd__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.YES),
    "__rsub__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rmul__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rmatmul__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rtruediv__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rfloordiv__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rmod__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rdivmod__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rpow__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rlshift__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rrshift__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rand__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__rxor__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__ror__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__iadd__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.YES),
    "__isub__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__imul__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__imatmul__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__itruediv__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__ifloordiv__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__imod__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__ipow__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__ilshift__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__irshift__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__iand__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__ixor__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__ior__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.MAYBE),
    "__neg__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__pos__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__abs__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__invert__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__complex__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__int__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__float__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__str__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__bytes__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__index__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__round__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__trunc__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__floor__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__ceil__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.MAYBE),
    "__enter__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__exit__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__buffer__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__release_buffer__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__hash__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__await__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__aiter__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__anext__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__aenter__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__aexit__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
    "__copy__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.YES),
    "__deepcopy__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.YES),
    "__replace__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.YES),
    "__format__": MethodInfo(state_changing=False, returns_same_type=YesNoMaybe.NO),
    "__delattr__": MethodInfo(state_changing=True, returns_same_type=YesNoMaybe.NO),
}

validate_cls_counts module-attribute 

validate_cls_counts: defaultdict[str, int] = defaultdict(int)

DoubleTypeVarStore

Bases: Generic[_T, _U]

METHOD DESCRIPTION
__init__
Source code in src/omnipy/data/helpers.py 
40
41
42
class DoubleTypeVarStore(Generic[_T, _U]):
    def __init__(self, t: _T | _U) -> None:
        raise ValueError()

__init__ 

__init__(t: _T | _U) -> None
Source code in src/omnipy/data/helpers.py 
41
42
def __init__(self, t: _T | _U) -> None:
    raise ValueError()

FailedData dataclass

METHOD DESCRIPTION
__init__
ATTRIBUTE DESCRIPTION
exception

TYPE: BaseException

job_name

TYPE: str

job_unique_name

TYPE: str

Source code in src/omnipy/data/helpers.py 
282
283
284
285
286
@dataclass(frozen=True, kw_only=True)
class FailedData:
    job_name: str
    job_unique_name: str = ''
    exception: BaseException

exception instance-attribute 

exception: BaseException

job_name instance-attribute 

job_name: str

job_unique_name class-attribute instance-attribute 

job_unique_name: str = ''

__init__ 

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

MethodInfo

Bases: NamedTuple

ATTRIBUTE DESCRIPTION
returns_same_type

TYPE: YesNoMaybe

state_changing

TYPE: bool

Source code in src/omnipy/data/helpers.py 
67
68
69
class MethodInfo(NamedTuple):
    state_changing: bool
    returns_same_type: YesNoMaybe

returns_same_type instance-attribute 

returns_same_type: YesNoMaybe

state_changing instance-attribute 

state_changing: bool

PendingData dataclass

METHOD DESCRIPTION
__init__
ATTRIBUTE DESCRIPTION
job_name

TYPE: str

job_unique_name

TYPE: str

Source code in src/omnipy/data/helpers.py 
276
277
278
279
@dataclass(frozen=True, kw_only=True)
class PendingData:
    job_name: str
    job_unique_name: str = ''

job_name instance-attribute 

job_name: str

job_unique_name class-attribute instance-attribute 

job_unique_name: str = ''

__init__ 

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

ResetSolutionTuple

Bases: NamedTuple

ATTRIBUTE DESCRIPTION
reset_solution

TYPE: ContextManager[None]

snapshot_taken

TYPE: bool

Source code in src/omnipy/data/helpers.py 
186
187
188
class ResetSolutionTuple(NamedTuple):
    reset_solution: ContextManager[None]
    snapshot_taken: bool

reset_solution instance-attribute 

reset_solution: ContextManager[None]

snapshot_taken instance-attribute 

snapshot_taken: bool

TypeVarStore

Bases: Generic[_T]

METHOD DESCRIPTION
__init__
Source code in src/omnipy/data/helpers.py 
35
36
37
class TypeVarStore(Generic[_T]):
    def __init__(self, t: _T) -> None:
        raise ValueError()

__init__ 

__init__(t: _T) -> None
Source code in src/omnipy/data/helpers.py 
36
37
def __init__(self, t: _T) -> None:
    raise ValueError()

TypeVarStore1

Bases: TypeVarStore[_T], Generic[_T]

METHOD DESCRIPTION
__init__
Source code in src/omnipy/data/helpers.py 
45
46
class TypeVarStore1(TypeVarStore[_T], Generic[_T]):
    ...

__init__ 

__init__(t: _T) -> None
Source code in src/omnipy/data/helpers.py 
36
37
def __init__(self, t: _T) -> None:
    raise ValueError()

TypeVarStore2

Bases: TypeVarStore[_T], Generic[_T]

METHOD DESCRIPTION
__init__
Source code in src/omnipy/data/helpers.py 
49
50
class TypeVarStore2(TypeVarStore[_T], Generic[_T]):
    ...

__init__ 

__init__(t: _T) -> None
Source code in src/omnipy/data/helpers.py 
36
37
def __init__(self, t: _T) -> None:
    raise ValueError()

TypeVarStore3

Bases: TypeVarStore[_T], Generic[_T]

METHOD DESCRIPTION
__init__
Source code in src/omnipy/data/helpers.py 
53
54
class TypeVarStore3(TypeVarStore[_T], Generic[_T]):
    ...

__init__ 

__init__(t: _T) -> None
Source code in src/omnipy/data/helpers.py 
36
37
def __init__(self, t: _T) -> None:
    raise ValueError()

TypeVarStore4

Bases: TypeVarStore[_T], Generic[_T]

METHOD DESCRIPTION
__init__
Source code in src/omnipy/data/helpers.py 
57
58
class TypeVarStore4(TypeVarStore[_T], Generic[_T]):
    ...

__init__ 

__init__(t: _T) -> None
Source code in src/omnipy/data/helpers.py 
36
37
def __init__(self, t: _T) -> None:
    raise ValueError()

YesNoMaybe

Bases: IntEnum

ATTRIBUTE DESCRIPTION
MAYBE

NO

YES

Source code in src/omnipy/data/helpers.py 
61
62
63
64
class YesNoMaybe(IntEnum):
    NO = 0
    YES = 1
    MAYBE = 2

MAYBE class-attribute instance-attribute 

MAYBE = 2

NO class-attribute instance-attribute 

NO = 0

YES class-attribute instance-attribute 

YES = 1

build_own_module_and_global_namespace_for_forward_refs 

build_own_module_and_global_namespace_for_forward_refs(
    own_class: type, calling_module: str | None, **localns: Any
) -> tuple[dict[str, Any], dict[str, Any]]

Build own-module and global namespaces for fwd reference resolution

Build global namespaces for forward reference resolution by merging:

  1. The model's own defining module namespace — so forward refs declared inside the model (e.g. JsonScalar in tables/models.py) are always resolvable even when update_forward_refs is triggered from a different module.

  2. The calling module's namespace on top — so forward refs that point to types available at the call site (e.g. NestedDataset in datasets.py) are resolved correctly.

  3. Any explicitly passed localns with the highest priority.

RETURNS DESCRIPTION
tuple[dict[str, Any], dict[str, Any]]

A tuple of (own_module_namespace, global_namespace) to be used for forward reference resolution.
Source code in src/omnipy/data/helpers.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def build_own_module_and_global_namespace_for_forward_refs(
    own_class: type,
    calling_module: str | None,
    **localns: Any,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Build own-module and global namespaces for fwd reference resolution

    Build global namespaces for forward reference resolution by merging:

    1. The model's own defining module namespace — so forward refs
       declared inside the model (e.g. JsonScalar in
       tables/models.py) are always resolvable even when
       update_forward_refs is triggered from a different module.

    2. The calling module's namespace on top — so forward refs
       that point to types available at the call site (e.g.
       NestedDataset in datasets.py) are resolved correctly.

    3. Any explicitly passed localns with the highest priority.

    Returns:
        A tuple of (own_module_namespace, global_namespace) to be used for
        forward reference resolution.
    """
    def _module_loaded(module_name: str | None) -> TypeIs[str]:
        return module_name is not None and module_name in sys.modules

    own_module = getattr(own_class, '__module__', None)
    own_module_ns = (sys.modules[own_module].__dict__ if _module_loaded(own_module) else {})

    calling_module_ns = (
        sys.modules[calling_module].__dict__ if _module_loaded(calling_module) else {})

    globalns: dict[str, Any] = {}
    globalns.update(own_module_ns)
    globalns.update(calling_module_ns)
    globalns.update(localns)

    return own_module_ns, globalns

cleanup_name_qualname_and_module 

cleanup_name_qualname_and_module(
    cls: type[DataClassBase], model_or_dataset: type[DataClassBase], orig_model: TypeForm
) -> None
Source code in src/omnipy/data/helpers.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def cleanup_name_qualname_and_module(
    cls: type[DataClassBase],
    model_or_dataset: type[DataClassBase],
    orig_model: TypeForm,
) -> None:
    def _display_as_type(model: TypeForm):
        if isinstance(model, str):  # ForwardRef
            return model
        elif isinstance(model, ForwardRef):
            return model.__forward_arg__
        elif isinstance(model, tuple):
            return ', '.join(_display_as_type(arg) for arg in model)
        elif is_union(model):
            return ' | '.join(_display_as_type(arg) for arg in get_args(model))
        elif len(get_args(model)) > 0:
            return (f'{_display_as_type(get_origin(model))}'
                    f"[{', '.join(_display_as_type(arg) for arg in get_args(model))}]")
        elif isinstance(model, TypeVar):
            return str(model)
        else:
            with suppress(AttributeError):
                return model.__name__  # type:ignore[union-attr]
            return str(model)

    params_str = _display_as_type(orig_model)

    model_or_dataset.__name__ = format_classname_with_params(cls.__name__, params_str)
    model_or_dataset.__qualname__ = format_classname_with_params(cls.__qualname__, params_str)
    model_or_dataset.__module__ = cls.__module__

debug_get_sorted_validate_counts 

debug_get_sorted_validate_counts() -> dict[str, int]
Source code in src/omnipy/data/helpers.py 
191
192
def debug_get_sorted_validate_counts() -> dict[str, int]:
    return dict(reversed(sorted(validate_cls_counts.items(), key=lambda item: item[1])))

debug_get_total_validate_count 

debug_get_total_validate_count() -> int
Source code in src/omnipy/data/helpers.py 
195
196
def debug_get_total_validate_count() -> int:
    return sum(val for key, val in validate_cls_counts.items())