Skip to content

omnipy.data.snapshot

Snapshot helpers that support change detection for Omnipy data objects.

CLASS DESCRIPTION
SnapshotHolder

Manage weakly referenced content snapshots for Omnipy models and datasets.
SnapshotWrapper

Store a snapshot together with the identity of the source object.
ATTRIBUTE DESCRIPTION
obj_getattr

obj_setattr

obj_getattr module-attribute 

obj_getattr = object.__getattribute__

obj_setattr module-attribute 

obj_setattr = object.__setattr__

SnapshotHolder

Bases: WeakKeyRefContainer[HasContentT, IsSnapshotWrapper[HasContentT, ContentT]], Generic[HasContentT, ContentT]


              flowchart BT
              omnipy.data.snapshot.SnapshotHolder[SnapshotHolder]
              omnipy.util.weak.WeakKeyRefContainer[WeakKeyRefContainer]

                              omnipy.util.weak.WeakKeyRefContainer --> omnipy.data.snapshot.SnapshotHolder
                


              click omnipy.data.snapshot.SnapshotHolder href "" "omnipy.data.snapshot.SnapshotHolder"
              click omnipy.util.weak.WeakKeyRefContainer href "" "omnipy.util.weak.WeakKeyRefContainer"

Manage weakly referenced content snapshots for Omnipy models and datasets.

The holder keeps snapshot copies alongside memoized deepcopy state so Omnipy can compare current content against prior state without permanently retaining deleted objects.

METHOD DESCRIPTION
__init__
all_are_empty

Return whether both snapshot storage and deepcopy bookkeeping are empty.
clear

Remove all stored snapshots and deepcopy bookkeeping state.
delete_scheduled_deepcopy_content_ids

Remove any queued content ids from the deepcopy memo.
get

Return the value associated with a live key object.
get_deepcopy_content_ids

Return the content ids currently retained by the deepcopy memo.
get_deepcopy_content_ids_scheduled_for_deletion

Return the queued content ids that should be pruned from the deepcopy memo.
in_deepcopy_content_ids

Return whether key is tracked in the deepcopy memo.
schedule_deepcopy_content_ids_for_deletion

Queue deepcopy-tracked content ids for deferred deletion when they are still known.
take_snapshot

Capture and store a snapshot of obj.content.
take_snapshot_setup

Disable garbage collection before a snapshot run begins.
take_snapshot_teardown

Prune stale deepcopy state and re-enable garbage collection after snapshotting.
Source code in src/omnipy/data/snapshot.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 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
class SnapshotHolder(WeakKeyRefContainer[HasContentT, IsSnapshotWrapper[HasContentT, ContentT]],
                     Generic[HasContentT, ContentT]):
    """Manage weakly referenced content snapshots for Omnipy models and datasets.

    The holder keeps snapshot copies alongside memoized deepcopy state so Omnipy can compare
    current content against prior state without permanently retaining deleted objects.
    """
    def __init__(self) -> None:
        super().__init__()
        self._deepcopy_memo = RefCountMemoDict[ContentT]()
        self._deepcopy_content_ids_for_deleted_objs: SetDeque[int] = SetDeque()

    def __setitem__(self, key: HasContentT, value: IsSnapshotWrapper[HasContentT,
                                                                     ContentT]) -> None:
        raise TypeError(f"'{self.__class__.__name__}' object does not support item assignment")

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

        self._deepcopy_content_ids_for_deleted_objs.clear()
        self._deepcopy_memo.clear()
        super().clear()

    def all_are_empty(self, debug: bool = False) -> bool:
        """Return whether both snapshot storage and deepcopy bookkeeping are empty.

        Args:
            debug: Whether to print detailed internal state before returning.

        Returns:
            ``True`` when no snapshots or pending deepcopy references remain.
        """

        _deepcopy_memo_all_are_empty = self._deepcopy_memo.all_are_empty(debug=debug)

        _all_are_empty = (
            len(self) == 0 and len(self._deepcopy_content_ids_for_deleted_objs) == 0
            and _deepcopy_memo_all_are_empty)

        if debug:
            print()
            print(f'SnapshotHolder.all_are_empty(): {_all_are_empty}')
            print('-------------------------')
            print(f'len(self): {len(self)}')
            print(f'self.get_deepcopy_content_ids(): '
                  f'{self.get_deepcopy_content_ids()}')
            print(f'self.get_deepcopy_content_ids_scheduled_for_deletion(): '
                  f'{self.get_deepcopy_content_ids_scheduled_for_deletion()}')
            print(f'self._deepcopy_memo.all_are_empty(): {_deepcopy_memo_all_are_empty}')

            if len(self) > 0:
                print('=========================')
                for key, val in self._value_dict.items():
                    print(f'{key}: {repr(val)}')

        return _all_are_empty

    def in_deepcopy_content_ids(self, key: int) -> bool:
        """Return whether ``key`` is tracked in the deepcopy memo."""

        return self._deepcopy_memo.in_deepcopy_object_ids(key)

    def get_deepcopy_content_ids(self) -> SetDeque[int]:
        """Return the content ids currently retained by the deepcopy memo."""

        return self._deepcopy_memo.get_deepcopy_object_ids()

    def get_deepcopy_content_ids_scheduled_for_deletion(self) -> SetDeque[int]:
        """Return the queued content ids that should be pruned from the deepcopy memo."""

        return self._deepcopy_content_ids_for_deleted_objs

    def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
        """Queue deepcopy-tracked content ids for deferred deletion when they are still known."""

        for key in keys:
            if self.in_deepcopy_content_ids(key):
                self._deepcopy_content_ids_for_deleted_objs.append(key)

    def delete_scheduled_deepcopy_content_ids(self) -> None:
        """Remove any queued content ids from the deepcopy memo."""

        keys_for_deleted_objs = self._deepcopy_content_ids_for_deleted_objs
        if len(keys_for_deleted_objs) > 0:
            # self._deepcopy_content_ids_for_deleted_objs = SetDeque()
            deepcopy_memo = obj_getattr(self, '_deepcopy_memo')
            deepcopy_memo.recursively_remove_deleted_objs(keys_for_deleted_objs)

    def take_snapshot_setup(self) -> None:
        """Disable garbage collection before a snapshot run begins."""

        gc.disable()

    def take_snapshot_teardown(self) -> None:
        """Prune stale deepcopy state and re-enable garbage collection after snapshotting."""

        self.delete_scheduled_deepcopy_content_ids()
        gc.enable()

    def take_snapshot(self, obj: HasContentT) -> None:
        """Capture and store a snapshot of ``obj.content``.

        The method first tries a memo-aware ``deepcopy`` so repeated snapshots can reuse preserved
        object fragments efficiently. If that fails, it falls back to plain ``deepcopy`` and then
        ``copy``.

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

        try:
            # Delete scheduled content in the deepcopy memo if the new object is reusing an old id.
            # This deletion might not succeed, e.g. if the current snapshot holds a reference to the
            # old object. In those case, setup_deepcopy() will raise an AssertionError, which should
            # trigger a new attempt to deepcopy without the memo dict.

            if self.in_deepcopy_content_ids(id(obj.content)):
                self.delete_scheduled_deepcopy_content_ids()

            obj_copy: ContentT

            with setup_and_teardown_callback_context(
                    setup_func=self._deepcopy_memo.setup_deepcopy,
                    setup_func_args=(obj.content,),
                    exception_func=self._deepcopy_memo.teardown_deepcopy,
                    teardown_func=self._deepcopy_memo.teardown_deepcopy,
            ):

                obj_copy = deepcopy(obj.content, self._deepcopy_memo)  # type: ignore[arg-type]
                self._deepcopy_memo.keep_alive_after_deepcopy()
        except (TypeError, ValueError, ValidationError, AssertionError) as exp:
            print(f'Error in deepcopy with memo dict: {exp}. '
                  f'Attempting deepcopy without memo dict.')
            try:
                # print(f'object content after retry: {obj.content}')
                obj_copy = deepcopy(obj.content)
            except (TypeError, ValueError, ValidationError, AssertionError) as exp:
                print(f'Error in deepcopy without memo dict: {exp}. '
                      f'Attempting simple copy.')
                obj_copy = copy(obj.content)

        # Eventual old snapshot object is being kept alive until this point, but is scheduled for
        # deletion after the next line. In many cases (but not all), this happens before
        # take_snapshot_teardown() is called, which triggers deletion of any unreferenced
        # fragments still kept alive in the memo dict.

        super().__setitem__(obj, SnapshotWrapper(id(obj), obj_copy))

__init__ 

__init__() -> None
Source code in src/omnipy/data/snapshot.py 
51
52
53
54
def __init__(self) -> None:
    super().__init__()
    self._deepcopy_memo = RefCountMemoDict[ContentT]()
    self._deepcopy_content_ids_for_deleted_objs: SetDeque[int] = SetDeque()

all_are_empty 

all_are_empty(debug: bool = False) -> bool

Return whether both snapshot storage and deepcopy bookkeeping are empty.

PARAMETER DESCRIPTION
debug

Whether to print detailed internal state before returning.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
bool

True when no snapshots or pending deepcopy references remain.
Source code in src/omnipy/data/snapshot.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def all_are_empty(self, debug: bool = False) -> bool:
    """Return whether both snapshot storage and deepcopy bookkeeping are empty.

    Args:
        debug: Whether to print detailed internal state before returning.

    Returns:
        ``True`` when no snapshots or pending deepcopy references remain.
    """

    _deepcopy_memo_all_are_empty = self._deepcopy_memo.all_are_empty(debug=debug)

    _all_are_empty = (
        len(self) == 0 and len(self._deepcopy_content_ids_for_deleted_objs) == 0
        and _deepcopy_memo_all_are_empty)

    if debug:
        print()
        print(f'SnapshotHolder.all_are_empty(): {_all_are_empty}')
        print('-------------------------')
        print(f'len(self): {len(self)}')
        print(f'self.get_deepcopy_content_ids(): '
              f'{self.get_deepcopy_content_ids()}')
        print(f'self.get_deepcopy_content_ids_scheduled_for_deletion(): '
              f'{self.get_deepcopy_content_ids_scheduled_for_deletion()}')
        print(f'self._deepcopy_memo.all_are_empty(): {_deepcopy_memo_all_are_empty}')

        if len(self) > 0:
            print('=========================')
            for key, val in self._value_dict.items():
                print(f'{key}: {repr(val)}')

    return _all_are_empty

clear 

clear() -> None

Remove all stored snapshots and deepcopy bookkeeping state.

Source code in src/omnipy/data/snapshot.py 
60
61
62
63
64
65
def clear(self) -> None:
    """Remove all stored snapshots and deepcopy bookkeeping state."""

    self._deepcopy_content_ids_for_deleted_objs.clear()
    self._deepcopy_memo.clear()
    super().clear()

delete_scheduled_deepcopy_content_ids 

delete_scheduled_deepcopy_content_ids() -> None

Remove any queued content ids from the deepcopy memo.

Source code in src/omnipy/data/snapshot.py 
123
124
125
126
127
128
129
130
def delete_scheduled_deepcopy_content_ids(self) -> None:
    """Remove any queued content ids from the deepcopy memo."""

    keys_for_deleted_objs = self._deepcopy_content_ids_for_deleted_objs
    if len(keys_for_deleted_objs) > 0:
        # self._deepcopy_content_ids_for_deleted_objs = SetDeque()
        deepcopy_memo = obj_getattr(self, '_deepcopy_memo')
        deepcopy_memo.recursively_remove_deleted_objs(keys_for_deleted_objs)

get 

get(key: _AnyKeyT) -> _ValT | None

Return the value associated with a live key object.

PARAMETER DESCRIPTION
key

Key object to look up by identity.

TYPE: _AnyKeyT

RETURNS DESCRIPTION
_ValT | None

The stored value for the key, or None when no entry exists.
Source code in src/omnipy/util/weak.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def get(self, key: _AnyKeyT) -> _ValT | None:
    """Return the value associated with a live key object.

    Args:
        key: Key object to look up by identity.

    Returns:
        The stored value for the key, or ``None`` when no entry exists.
    """
    key_ref = KeyRef(key)
    if key_ref in self._value_dict:
        return self._value_dict[key_ref]
    else:
        return None

get_deepcopy_content_ids 

get_deepcopy_content_ids() -> SetDeque[int]

Return the content ids currently retained by the deepcopy memo.

Source code in src/omnipy/data/snapshot.py 
106
107
108
109
def get_deepcopy_content_ids(self) -> SetDeque[int]:
    """Return the content ids currently retained by the deepcopy memo."""

    return self._deepcopy_memo.get_deepcopy_object_ids()

get_deepcopy_content_ids_scheduled_for_deletion 

get_deepcopy_content_ids_scheduled_for_deletion() -> SetDeque[int]

Return the queued content ids that should be pruned from the deepcopy memo.

Source code in src/omnipy/data/snapshot.py 
111
112
113
114
def get_deepcopy_content_ids_scheduled_for_deletion(self) -> SetDeque[int]:
    """Return the queued content ids that should be pruned from the deepcopy memo."""

    return self._deepcopy_content_ids_for_deleted_objs

in_deepcopy_content_ids 

in_deepcopy_content_ids(key: int) -> bool

Return whether key is tracked in the deepcopy memo.

Source code in src/omnipy/data/snapshot.py 
101
102
103
104
def in_deepcopy_content_ids(self, key: int) -> bool:
    """Return whether ``key`` is tracked in the deepcopy memo."""

    return self._deepcopy_memo.in_deepcopy_object_ids(key)

schedule_deepcopy_content_ids_for_deletion 

schedule_deepcopy_content_ids_for_deletion(*keys: int) -> None

Queue deepcopy-tracked content ids for deferred deletion when they are still known.

Source code in src/omnipy/data/snapshot.py 
116
117
118
119
120
121
def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
    """Queue deepcopy-tracked content ids for deferred deletion when they are still known."""

    for key in keys:
        if self.in_deepcopy_content_ids(key):
            self._deepcopy_content_ids_for_deleted_objs.append(key)

take_snapshot 

take_snapshot(obj: HasContentT) -> None

Capture and store a snapshot of obj.content.

The method first tries a memo-aware deepcopy so repeated snapshots can reuse preserved object fragments efficiently. If that fails, it falls back to plain deepcopy and then copy.

PARAMETER DESCRIPTION
obj

Object whose content attribute should be snapshotted.

TYPE: HasContentT

Source code in src/omnipy/data/snapshot.py 
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
def take_snapshot(self, obj: HasContentT) -> None:
    """Capture and store a snapshot of ``obj.content``.

    The method first tries a memo-aware ``deepcopy`` so repeated snapshots can reuse preserved
    object fragments efficiently. If that fails, it falls back to plain ``deepcopy`` and then
    ``copy``.

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

    try:
        # Delete scheduled content in the deepcopy memo if the new object is reusing an old id.
        # This deletion might not succeed, e.g. if the current snapshot holds a reference to the
        # old object. In those case, setup_deepcopy() will raise an AssertionError, which should
        # trigger a new attempt to deepcopy without the memo dict.

        if self.in_deepcopy_content_ids(id(obj.content)):
            self.delete_scheduled_deepcopy_content_ids()

        obj_copy: ContentT

        with setup_and_teardown_callback_context(
                setup_func=self._deepcopy_memo.setup_deepcopy,
                setup_func_args=(obj.content,),
                exception_func=self._deepcopy_memo.teardown_deepcopy,
                teardown_func=self._deepcopy_memo.teardown_deepcopy,
        ):

            obj_copy = deepcopy(obj.content, self._deepcopy_memo)  # type: ignore[arg-type]
            self._deepcopy_memo.keep_alive_after_deepcopy()
    except (TypeError, ValueError, ValidationError, AssertionError) as exp:
        print(f'Error in deepcopy with memo dict: {exp}. '
              f'Attempting deepcopy without memo dict.')
        try:
            # print(f'object content after retry: {obj.content}')
            obj_copy = deepcopy(obj.content)
        except (TypeError, ValueError, ValidationError, AssertionError) as exp:
            print(f'Error in deepcopy without memo dict: {exp}. '
                  f'Attempting simple copy.')
            obj_copy = copy(obj.content)

    # Eventual old snapshot object is being kept alive until this point, but is scheduled for
    # deletion after the next line. In many cases (but not all), this happens before
    # take_snapshot_teardown() is called, which triggers deletion of any unreferenced
    # fragments still kept alive in the memo dict.

    super().__setitem__(obj, SnapshotWrapper(id(obj), obj_copy))

take_snapshot_setup 

take_snapshot_setup() -> None

Disable garbage collection before a snapshot run begins.

Source code in src/omnipy/data/snapshot.py 
132
133
134
135
def take_snapshot_setup(self) -> None:
    """Disable garbage collection before a snapshot run begins."""

    gc.disable()

take_snapshot_teardown 

take_snapshot_teardown() -> None

Prune stale deepcopy state and re-enable garbage collection after snapshotting.

Source code in src/omnipy/data/snapshot.py 
137
138
139
140
141
def take_snapshot_teardown(self) -> None:
    """Prune stale deepcopy state and re-enable garbage collection after snapshotting."""

    self.delete_scheduled_deepcopy_content_ids()
    gc.enable()

SnapshotWrapper dataclass

Bases: Generic[ObjContraT, ContentT]


              flowchart BT
              omnipy.data.snapshot.SnapshotWrapper[SnapshotWrapper]

              

              click omnipy.data.snapshot.SnapshotWrapper href "" "omnipy.data.snapshot.SnapshotWrapper"

Store a snapshot together with the identity of the source object.

ATTRIBUTE DESCRIPTION
id

id() of the object the snapshot was taken from.

TYPE: int

snapshot

Copied content captured from that object.

TYPE: ContentT

METHOD DESCRIPTION
__init__
differs_from

Return whether obj currently differs from the stored snapshot.
taken_of_same_obj

Return whether this snapshot was taken from obj itself.
Source code in src/omnipy/data/snapshot.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass
class SnapshotWrapper(Generic[ObjContraT, ContentT]):
    """Store a snapshot together with the identity of the source object.

    Attributes:
        id: ``id()`` of the object the snapshot was taken from.
        snapshot: Copied content captured from that object.
    """

    id: int
    snapshot: ContentT

    def taken_of_same_obj(self, obj: ObjContraT) -> bool:
        """Return whether this snapshot was taken from ``obj`` itself."""

        return self.id == id(obj)

    def differs_from(self, obj: ObjContraT) -> bool:
        """Return whether ``obj`` currently differs from the stored snapshot."""

        return not all_equals(self.snapshot, obj)

id instance-attribute 

id: int

snapshot instance-attribute 

snapshot: ContentT

__init__ 

__init__(id: int, snapshot: ContentT) -> None

differs_from 

differs_from(obj: ObjContraT) -> bool

Return whether obj currently differs from the stored snapshot.

Source code in src/omnipy/data/snapshot.py 
34
35
36
37
def differs_from(self, obj: ObjContraT) -> bool:
    """Return whether ``obj`` currently differs from the stored snapshot."""

    return not all_equals(self.snapshot, obj)

taken_of_same_obj 

taken_of_same_obj(obj: ObjContraT) -> bool

Return whether this snapshot was taken from obj itself.

Source code in src/omnipy/data/snapshot.py 
29
30
31
32
def taken_of_same_obj(self, obj: ObjContraT) -> bool:
    """Return whether this snapshot was taken from ``obj`` itself."""

    return self.id == id(obj)