Mostly working refactored snapshot functionality. Failing tests are (…

…mostly) unrelated

pyproject.toml

@@ -43,6 +43,7 @@ humanize = "^4.9.0"
 httpx = "^0.26.0"
 pydantic = {version = "<2", extras = ["email"]}
 bidict = "^0.23.1"
+line-profiler-pycharm = "^1.1.0"
 
 [tool.poetry.group.dev.dependencies]
 deepdiff = "^6.2.1"

scripts/deepcopy_memo_frozen.py

@@ -0,0 +1,61 @@
+from typing import TypeAlias
+
+from omnipy.modules.general.models import Model, NestedFrozenDictsOrTuplesModel
+from tests.modules.general.cases.raw.examples import (e_complex_key_dict,
+                                                      e_int_key_dict,
+                                                      e_none_key_dict,
+                                                      ej_frozendict_iterable_scalar,
+                                                      ej_frozendict_iterable_scalar_empty,
+                                                      ej_frozendict_wrong_scalar,
+                                                      ej_tuple_iterable_scalar,
+                                                      ej_tuple_wrong_scalar,
+                                                      ej_type,
+                                                      f_complex,
+                                                      f_dict,
+                                                      f_frozendict,
+                                                      f_int,
+                                                      f_list,
+                                                      f_none,
+                                                      f_set,
+                                                      f_str,
+                                                      f_tuple)
+
+FSK: TypeAlias = int | str | complex  # for keys
+FSV: TypeAlias = None | int | str | complex  # for values
+
+
+def run_nested_frozen_dicts_or_tuples_model() -> None:
+    _two_level_list: list[FSV | list[FSV] | dict[str, FSV]] = \
+            f_list + [list(f_list)] + [dict(f_dict)]
+    _two_level_dict: dict[str, str | list[FSV] | dict[str, FSV]] = \
+            {'a': f_str, 'b': list(f_list), 'c': dict(f_dict)}
+
+    number_model = Model[int](123)
+
+    print(number_model.snapshot_holder._deepcopy_memo)
+
+    number_model.snapshot_holder._deepcopy_memo.clear()
+
+    print(number_model.snapshot_holder._deepcopy_memo)
+
+    nested_frozen_dicts_or_tuples = (
+        list(f_list + [list(f_list), dict(f_dict), list(_two_level_list), dict(_two_level_dict)]))
+    nested_frozen_dicts_or_tuples_model = NestedFrozenDictsOrTuplesModel(
+        nested_frozen_dicts_or_tuples)
+
+    print(len(number_model.snapshot_holder._deepcopy_memo))
+    k = 0
+    v = 0
+    for k, v in number_model.snapshot_holder._deepcopy_memo.items():
+        print(f'{k}: {v}')
+    del k
+    del v
+
+    del nested_frozen_dicts_or_tuples_model
+    number_model.snapshot_holder.delete_scheduled()
+    print(number_model.snapshot_holder._deepcopy_memo)
+
+    assert len(number_model.snapshot_holder._deepcopy_memo) == 0
+
+
+run_nested_frozen_dicts_or_tuples_model()

scripts/profile_module_as_script.py

@@ -0,0 +1,32 @@
+import cProfile
+from importlib import import_module
+import os
+import pstats
+import sys
+
+if __name__ == '__main__':
+    script_path = os.path.abspath(sys.argv[1])
+    project_path = os.path.abspath(__file__ + '/../..')
+
+    print(f'script_path: {script_path}')
+    print(f'project_path: {project_path}')
+
+    assert script_path.startswith(project_path)
+    module = '.'.join(script_path.split('.')[0][len(project_path) + 1:].split('/'))
+
+    # print(f'package: {package}')
+    print(f'module: {module}')
+
+    sys.path.append(project_path)
+
+    mod = import_module(module)
+
+    profiler = cProfile.Profile()
+    profiler.enable()
+
+    mod.run()
+
+    profiler.disable()
+    stats = pstats.Stats(profiler).sort_stats('cumtime')
+    stats.dump_stats('latest4.pstats')
+    print('dumped')

src/omnipy/api/protocols/private/data.py

@@ -1,4 +1,4 @@
-from typing import Protocol, TypeVar
+from typing import Callable, ContextManager, Protocol, TypeVar
 
 from omnipy.api.protocols.private.util import IsSnapshotHolder
 from omnipy.api.protocols.public.config import IsDataConfig
@@ -20,6 +20,13 @@ def set_config(self, config: IsDataConfig) -> None:
     def snapshot_holder(self) -> IsSnapshotHolder[_ObjT, _ContentsT]:
         ...
 
+    def deepcopy_context(
+        self,
+        top_level_entry_func: Callable[[], None],
+        top_level_exit_func: Callable[[], None],
+    ) -> ContextManager[int]:
+        ...
+
 
 class IsDataClassBase(Protocol):
     """"""

src/omnipy/api/protocols/private/util.py

@@ -9,6 +9,7 @@
 _ContentsT = TypeVar("_ContentsT", bound=object)
 _ContentCovT = TypeVar("_ContentCovT", covariant=True, bound=object)
 _ContentContraT = TypeVar("_ContentContraT", contravariant=True, bound=object)
+_HasContentsT = TypeVar('_HasContentsT', bound='HasContents')
 
 
 @runtime_checkable
@@ -25,6 +26,7 @@ def __call__(self, *args: object, **kwargs: object) -> Callable[[Callable], Deco
         ...
 
 
+@runtime_checkable
 class HasContents(Protocol[_ContentsT]):
     @property
     def contents(self) -> _ContentsT:
@@ -52,9 +54,9 @@ def __len__(self) -> int:
         ...
 
 
-class IsSnapshot(Protocol[_ObjContraT, _ContentsT]):
+class IsSnapshotWrapper(Protocol[_ObjContraT, _ContentsT]):
     id: int
-    obj_copy: _ContentsT
+    snapshot: _ContentsT
 
     def taken_of_same_obj(self, obj: _ObjContraT) -> bool:
         ...
@@ -63,14 +65,26 @@ def differs_from(self, obj: _ObjContraT) -> bool:
         ...
 
 
-class IsSnapshotHolder(IsWeakKeyRefContainer[HasContents[_ContentsT], IsSnapshot[_ObjT,
-                                                                                 _ContentsT]],
-                       Protocol[_ObjT, _ContentsT]):
+class IsSnapshotHolder(IsWeakKeyRefContainer[_HasContentsT,
+                                             IsSnapshotWrapper[_HasContentsT, _ContentsT]],
+                       Protocol[_HasContentsT, _ContentsT]):
     """"""
+    def schedule_for_deletion(self, key: int) -> None:
+        ...
+
+    def delete_scheduled(self) -> None:
+        ...
+
     def clear(self) -> None:
         ...
 
-    def take_snapshot(self, obj: HasContents[_ContentsT]) -> None:
+    def take_snapshot_setup(self) -> None:
+        ...
+
+    def take_snapshot_cleanup(self) -> None:
+        ...
+
+    def take_snapshot(self, obj: _HasContentsT) -> None:
         ...
 
     #

src/omnipy/data/data_class_creator.py

@@ -1,5 +1,6 @@
 from abc import ABCMeta
-from typing import Generic, TypeVar
+from contextlib import contextmanager
+from typing import Callable, ContextManager, Generic, Iterator, TypeVar
 
 from omnipy.api.protocols.private.data import IsDataClassCreator
 from omnipy.api.protocols.private.util import IsSnapshotHolder
@@ -12,6 +13,7 @@ class DataClassCreator:
     def __init__(self) -> None:
         self._config: IsDataConfig = DataConfig()
         self._snapshot_holder = SnapshotHolder[object, object]()
+        self._deepcopy_context_level = 0
 
     def set_config(self, config: IsDataConfig) -> None:
         self._config = config
@@ -24,6 +26,28 @@ def config(self) -> IsDataConfig:
     def snapshot_holder(self) -> IsSnapshotHolder[object, object]:
         return self._snapshot_holder
 
+    def deepcopy_context(
+        self,
+        top_level_entry_func: Callable[[], None],
+        top_level_exit_func: Callable[[], None],
+    ) -> ContextManager[int]:
+        @contextmanager
+        def _call_exit_func_if_top_level(*args, **kwds) -> Iterator[int]:
+            if self._deepcopy_context_level == 0:
+                top_level_entry_func()
+
+            self._deepcopy_context_level += 1
+
+            try:
+                yield self._deepcopy_context_level
+            finally:
+                self._deepcopy_context_level -= 1
+
+                if self._deepcopy_context_level == 0:
+                    top_level_exit_func()
+
+        return _call_exit_func_if_top_level()
+
 
 class DataClassBaseMeta(ABCMeta):
     """"""
@@ -46,3 +70,11 @@ def config(self) -> IsDataConfig:
     @property
     def snapshot_holder(self) -> IsSnapshotHolder:
         return self.__class__.data_class_creator.snapshot_holder
+
+    def deepcopy_context(
+        self,
+        top_level_entry_func: Callable[[], None],
+        top_level_exit_func: Callable[[], None],
+    ) -> ContextManager[int]:
+        return self.__class__.data_class_creator.deepcopy_context(top_level_entry_func,
+                                                                  top_level_exit_func)

src/omnipy/data/model.py

@@ -36,7 +36,7 @@
 from pydantic.utils import lenient_isinstance, lenient_issubclass
 
 from omnipy.api.exceptions import ParamException
-from omnipy.api.protocols.private.util import IsSnapshot, IsSnapshotHolder
+from omnipy.api.protocols.private.util import IsSnapshotHolder, IsSnapshotWrapper
 from omnipy.api.typedefs import TypeForm
 from omnipy.data.data_class_creator import DataClassBase, DataClassBaseMeta
 from omnipy.data.methodinfo import MethodInfo, SPECIAL_METHODS_INFO
@@ -58,7 +58,7 @@
                                  remove_annotated_plus_optional_if_present,
                                  remove_forward_ref_notation,
                                  RestorableContents,
-                                 Snapshot)
+                                 SnapshotWrapper)
 from omnipy.util.tabulate import tabulate
 
 _KeyT = TypeVar('_KeyT', bound=Hashable)
@@ -396,7 +396,7 @@ def __init__(
             else:
                 raise
 
-        self._take_snapshot_of_validated_contents()
+        # self._take_snapshot_of_validated_contents()
 
         if not self.__class__.__doc__:
             self._set_standard_field_description()
@@ -408,15 +408,15 @@ def _init(self, super_kwargs: dict[str, Any], **kwargs: Any) -> None:
 
     @staticmethod
     def _finalize(self_id: int, contents_id: int, snapshot_holder: IsSnapshotHolder):
-        print(f'Deleting self.id: {self_id} -> contents_id: {contents_id}')
-        snapshot_holder.keys_for_deleted_objs.append(contents_id)
+        # print(f'Deleting self.id: {self_id} -> contents_id: {contents_id}')
+        snapshot_holder.schedule_for_deletion(contents_id)
 
     # def __del__(self):
     #     # if self in self.snapshot_holder:
     #     print(f'Deleting {id(self)} -> {id(self.contents)}')
     #     contents_id = id(self.contents)
     #     self.contents = Undefined
-    #     self.snapshot_holder.keys_for_deleted_objs.append(contents_id)
+    #     self.snapshot_holder.schedule_for_deletion(contents_id)
 
     # if id(self) in _restorable_content_cache:
     #     del _restorable_content_cache[id(self)]
@@ -469,7 +469,9 @@ def validate_contents(self, restore_snapshot_if_interactive_and_invalid: bool =
     def _validate_and_set_value(self,
                                 new_contents: object,
                                 restore_snapshot_if_interactive_and_invalid: bool = False) -> None:
-        if restore_snapshot_if_interactive_and_invalid and self.config.interactive_mode:
+        if restore_snapshot_if_interactive_and_invalid \
+                and self.config.interactive_mode \
+                and self.has_snapshot():
             reset_solution = AttribHolder(self, 'contents', self.snapshot, reset_to_other=True)
         else:
             reset_solution = nothing()
@@ -494,30 +496,38 @@ def _get_restorable_contents(self):
 
     @property
     def snapshot(self) -> _RootT:
-        snapshot: IsSnapshot['Model', _RootT] = self.snapshot_holder[self]
-        assert snapshot.id == id(self)
-        return snapshot.obj_copy
+        snapshot_wrapper = self._get_snapshot_wrapper()
+        assert snapshot_wrapper.id == id(self)
+        return snapshot_wrapper.snapshot
 
     def has_snapshot(self) -> bool:
         return self in self.snapshot_holder
 
+    def _get_snapshot_wrapper(self) -> IsSnapshotWrapper['Model', _RootT]:
+        assert self.has_snapshot(), 'No snapshot taken yet'
+        return self.snapshot_holder[self]
+
     def snapshot_taken_of_same_model(self, model: 'Model') -> bool:
-        return self.snapshot.taken_of_same_obj(model)
+        snapshot_wrapper = self._get_snapshot_wrapper()
+        return snapshot_wrapper.taken_of_same_obj(model)
 
     def snapshot_differs_from_model(self, model: 'Model') -> bool:
-        return self.snapshot.differs_from(model.contents)
+        snapshot_wrapper = self._get_snapshot_wrapper()
+        return snapshot_wrapper.differs_from(model.contents)
 
-    @property
-    def contents_validated(self) -> bool:
+    def contents_validated_according_to_snapshot(self) -> bool:
         needs_validation = self.snapshot_differs_from_model(self) \
                            or not self.snapshot_taken_of_same_model(self)
         return not needs_validation
 
     def _take_snapshot_of_validated_contents(self) -> None:
         if self.config.interactive_mode:
-            self.snapshot_holder.take_snapshot(self)
-            print(
-                f'Snapshot contents_id={id(self.contents)} -> {id(self.snapshot)}: {self.contents}')
+            with self.deepcopy_context(self.snapshot_holder.take_snapshot_setup,
+                                       self.snapshot_holder.take_snapshot_cleanup):
+                self.snapshot_holder.take_snapshot(self)
+            # print(
+            #     f'SnapshotWrapper contents_id={id(self.contents)} -> {id(self.snapshot)}: {self.contents}'
+            # )
 
     @classmethod
     def _parse_data(cls, data: Any) -> _RootT:
@@ -716,14 +726,14 @@ def __setattr__(self, attr: str, value: Any) -> None:
     def _special_method(  # noqa: C901
             self, name: str, info: MethodInfo, *args: object, **kwargs: object) -> object:
 
-        if info.state_changing and self.config.interactive_mode:
-            if not self.contents_validated:
+        if info.state_changing and self.config.interactive_mode and self.has_snapshot():
+            if not self.contents_validated_according_to_snapshot():
                 self.validate_contents(restore_snapshot_if_interactive_and_invalid=True)
 
             reset_contents_to_prev = AttribHolder(self, 'contents', copy_attr=True)
             with reset_contents_to_prev:
                 ret = self._call_special_method(name, *args, **kwargs)
-                if self.snapshot_differs_from_model(self.contents):
+                if self.snapshot_differs_from_model(self):
                     self.validate_contents()
 
         elif name == '__iter__' and isinstance(self, Iterable):
@@ -901,10 +911,11 @@ def __eq__(self, other: object) -> bool:
             and self.to_data() == cast(Model, other).to_data()  # last line is just in case
 
     def __repr__(self) -> str:
-        if self.config.interactive_mode and not _waiting_for_terminal_repr():
-            if get_calling_module_name() in INTERACTIVE_MODULES:
-                _waiting_for_terminal_repr(True)
-                return self._table_repr()
+        # if self.config.interactive_mode and not _waiting_for_terminal_repr():
+        #     if get_calling_module_name() in INTERACTIVE_MODULES:
+        #         _waiting_for_terminal_repr(True)
+        #         return self._table_repr()
+        return self._trad_repr()
         return self._trad_repr()
 
     def view(self):