Skip to content

omnipy.util.memo

Reference-count-aware memo helpers for controlled deep-copy cleanup.

This module provides a custom memo dictionary used during deepcopy operations to track copied object ids, keep selected sub-objects alive temporarily, and prune memoized entries once their source objects are no longer referenced.

CLASS DESCRIPTION
RefCountMemoDict

Track deepcopy memo entries and their transitive object references.

RefCountMemoDict

Bases: UserDict[int, _ObjT], Generic[_ObjT]


              flowchart BT
              omnipy.util.memo.RefCountMemoDict[RefCountMemoDict]

              

              click omnipy.util.memo.RefCountMemoDict href "" "omnipy.util.memo.RefCountMemoDict"

Track deepcopy memo entries and their transitive object references.

The container stores non-atomic memoized objects by id, records which sub-object ids were touched while deep-copying a root object, and supports recursive cleanup of entries whose originals have been deleted.

ATTRIBUTE DESCRIPTION
_cur_deepcopy_obj_id

Id for the root object currently being deep-copied, or None when no deepcopy is active.

TYPE: int | None

_cur_keep_alive_list

Temporary keep-alive list used by deepcopy internals.

TYPE: list[_ObjT | None]

_keep_alive_dict

Persistent keep-alive storage keyed by object id.

TYPE: dict[int, _ObjT | None]

_sub_obj_ids

Mapping from root object id to the ordered ids of sub-objects touched while copying that root.

TYPE: defaultdict[int, SetDeque[int]]

METHOD DESCRIPTION
__init__
all_are_empty

Check whether all memo and bookkeeping structures are empty.
clear

Clear memoized entries and all internal bookkeeping maps.
get_deepcopy_object_ids

Return all registered deepcopy root object ids.
in_deepcopy_object_ids

Check whether an object id is registered as a deepcopy root.
keep_alive_after_deepcopy

Persist objects captured by deepcopy keep-alive bookkeeping.
recursively_remove_deleted_objs

Recursively remove memo entries whose source objects were deleted.
setup_deepcopy

Initialize bookkeeping for a new root-object deepcopy.
teardown_deepcopy

Finalize bookkeeping state after a deepcopy attempt.
Source code in src/omnipy/util/memo.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 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
191
192
193
194
195
196
197
198
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
228
229
230
231
232
233
234
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
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
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
class RefCountMemoDict(UserDict[int, _ObjT], Generic[_ObjT]):
    """Track deepcopy memo entries and their transitive object references.

    The container stores non-atomic memoized objects by id, records which
    sub-object ids were touched while deep-copying a root object, and supports
    recursive cleanup of entries whose originals have been deleted.

    Attributes:
        _cur_deepcopy_obj_id: Id for the root object currently being deep-copied,
            or ``None`` when no deepcopy is active.
        _cur_keep_alive_list: Temporary keep-alive list used by deepcopy internals.
        _keep_alive_dict: Persistent keep-alive storage keyed by object id.
        _sub_obj_ids: Mapping from root object id to the ordered ids of sub-objects
            touched while copying that root.
    """
    def __init__(self) -> None:
        super().__init__()
        self._cur_deepcopy_obj_id: int | None = None
        self._cur_keep_alive_list: list[_ObjT | None] = []
        self._keep_alive_dict: dict[int, _ObjT | None] = {}
        self._sub_obj_ids: defaultdict[int, SetDeque[int]] = defaultdict(SetDeque)

    def all_are_empty(self, debug: bool = False) -> bool:
        """Check whether all memo and bookkeeping structures are empty.

        Args:
            debug: When ``True``, print a detailed diagnostic report of current
                internal state and object referrers.

        Returns:
            ``True`` when memo data, keep-alive state, and active-copy state are
            all empty; otherwise ``False``.
        """
        _all_are_empty = (
            len(self) == 0 and len(self._keep_alive_dict) == 0 and len(self._sub_obj_ids) == 0
            and self._cur_deepcopy_obj_id is None and len(self._cur_keep_alive_list) == 0)

        if debug:
            print()
            print(f'RefCountMemoDict.all_are_empty(): {_all_are_empty}')
            print('--------------------------------------------------')
            print(f'len(self): {len(self)}')
            print(f'len(self._keep_alive_dict): {len(self._keep_alive_dict)}')
            print(f'len(self._sub_obj_ids): {len(self._sub_obj_ids)}')
            print(f'self._cur_deepcopy_obj_id: {self._cur_deepcopy_obj_id}')
            print(f'len(self._cur_keep_alive_list): {len(self._cur_keep_alive_list)}')

            if len(self) > 0:
                print('==================================================')
                for key in self:
                    print(f'Content id: {key} '
                          f'[{sys.getrefcount(self[key]) - 1} refs, id={id(self[key])}]: '
                          f'{repr(self[key])}')

                for key in self:
                    print('--------------------------------------------------')
                    print(f'References for {key}: {repr(self[key])}')
                    print('--------------------------------------------------')
                    ref = None
                    for i, ref in enumerate(gc.get_referrers(self[key])):
                        try:
                            print(f'[Reference {i}]')
                            print(f'    Type: {type(ref)}')
                            print(f'    ID: {id(ref)}')
                            print(f'    Value: {repr(ref)}')
                            print()

                        except Exception as e:
                            print(f'Error: {repr(e)}')
                            pass
                    if ref is not None:
                        del ref

        return _all_are_empty

    def clear(self):
        """Clear memoized entries and all internal bookkeeping maps."""
        super().clear()
        self._keep_alive_dict.clear()
        self._sub_obj_ids.clear()

    def in_deepcopy_object_ids(self, key: int) -> bool:
        """Check whether an object id is registered as a deepcopy root.

        Args:
            key: Object id to check.

        Returns:
            ``True`` when the id exists in the deepcopy root registry.
        """
        return key in self._sub_obj_ids

    def get_deepcopy_object_ids(self) -> SetDeque[int]:
        """Return all registered deepcopy root object ids.

        Returns:
            Object ids currently tracked as deepcopy roots, in insertion order.
        """
        return SetDeque(self._sub_obj_ids.keys())

    def setup_deepcopy(self, obj):
        """Initialize bookkeeping for a new root-object deepcopy.

        Args:
            obj: Root object about to be deep-copied.

        Raises:
            AssertionError: If a deepcopy is already in progress, temporary
                keep-alive state is not empty, or this object id is already
                registered in the memo state.
        """
        assert self._cur_deepcopy_obj_id is None, \
            f'self._cur_deepcopy_obj_id is not None, but {self._cur_deepcopy_obj_id}'
        assert len(self._cur_keep_alive_list) == 0, \
            f'len(self._cur_keep_alive_list) == {len(self._cur_keep_alive_list)} instead of 0'

        # Solution to test_ref_count_memo_dict_repeated_deepcopy_same_obj(), but not
        # necessary for the current implementation (see comments on the test). Fix unnecessarily
        # deletes shared fragments between the old and new deepcopies, and is thus disabled.
        # Instead, there is a check to ensure that the current object has not already been
        # deepcopied, which is the assumption behind disabling the fix

        assert id(obj) not in self._sub_obj_ids, \
            (f'Object with id: {id(obj)} already in deepcopy_memo. It has either already been '
             f'deepcopied, or it is reusing the id of a deleted object which has not been deleted '
             'from the deepcopy memo.')

        self._cur_deepcopy_obj_id = id(obj)

    def keep_alive_after_deepcopy(self):
        """Persist objects captured by deepcopy keep-alive bookkeeping.

        Non-atomic objects are retained directly in ``_keep_alive_dict`` while
        atomic objects are represented by ``None`` placeholders.
        """
        # old_sub_obj_ids = self._sub_obj_ids[self._cur_deepcopy_obj_id]
        # assert len(old_sub_obj_ids) == len(self._cur_keep_alive_list)
        #
        # self._sub_obj_ids[self._cur_deepcopy_obj_id] = SetDeque()

        while len(self._cur_keep_alive_list) > 0:
            keep_alive_obj = self._cur_keep_alive_list.pop()

            # id_keep_alive_obj = id(keep_alive_obj)
            # assert id_keep_alive_obj in old_sub_obj_ids
            # self._sub_obj_ids[self._cur_deepcopy_obj_id].append(id_keep_alive_obj)

            if not self._is_atomic(keep_alive_obj):
                self._keep_alive_dict[id(keep_alive_obj)] = keep_alive_obj
            else:
                self._keep_alive_dict[id(keep_alive_obj)] = None

    def teardown_deepcopy(self):
        """Finalize bookkeeping state after a deepcopy attempt.

        This clears active-copy markers and removes root-id tracking if the
        current root was not preserved in keep-alive storage.
        """
        # Also seems to be unnecessary now. This is for recovering from exceptions during deepcopy
        # that would leave the memo dict in an inconsistent state. This should now be handled by
        # SnapshotHolder.take_snapshot().
        #
        # for possibly_added_obj in self._sub_obj_ids[self._cur_deepcopy_obj_id]:
        #     if possibly_added_obj in self and possibly_added_obj not in self._keep_alive_dict:
        #         del self[possibly_added_obj]

        if self._cur_deepcopy_obj_id and \
                (self._cur_deepcopy_obj_id in self._sub_obj_ids
                 and self._cur_deepcopy_obj_id not in self._keep_alive_dict):
            del self._sub_obj_ids[self._cur_deepcopy_obj_id]

        self._cur_deepcopy_obj_id = None
        self._cur_keep_alive_list = []

    @staticmethod
    def _is_atomic(obj: object) -> bool:
        try:
            cls = type(obj)

            if cls is tuple:
                return True

            if sys.version_info >= (3, 14):
                return cls in _atomic_types
            else:
                return _deepcopy_dispatch[cls] is _deepcopy_atomic
        except KeyError:
            return False

    def __setitem__(self, key, obj):
        # try:
        #     print(f'{key},{id(obj)}: {obj} [{type(obj)}]')
        # except AttributeError:
        #     print(f'{key},{id(obj)}: [{type(obj)}]')
        #
        # if isinstance(obj, dict):
        #     print({k: id(v) for k, v in obj.items()})

        if key != id(self):
            if not self._is_atomic(obj):
                self.data[key] = obj
            self._register_key_as_sub_obj_of_cur_deepcopy_obj(key)

    def __getitem__(self, key: int) -> _ObjT | list[_ObjT | None]:  # type: ignore[override]
        if key == id(self):
            return self._cur_keep_alive_list
        else:
            ret = super().__getitem__(key)
            # print(f'Getting {key}')
            self._register_key_as_sub_obj_of_cur_deepcopy_obj(key)
            return ret

    def _register_key_as_sub_obj_of_cur_deepcopy_obj(self, key: int) -> None:
        if self._cur_deepcopy_obj_id is not None:
            self._sub_obj_ids[self._cur_deepcopy_obj_id].append(key)

    def recursively_remove_deleted_objs(
        self,
        keys_for_deleted_objs: SetDeque[int],
    ):
        """Recursively remove memo entries whose source objects were deleted.

        Args:
            keys_for_deleted_objs: Initial candidate object ids for deletion. The
                deque is updated in place with ids that could not be removed in the
                current pass.

        Raises:
            Exception: Re-raises unexpected exceptions from internal removal logic
                after printing debugging details.
        """
        try:
            # print(f'keys_for_deleted_objs: {keys_for_deleted_objs}')
            # print(f'self.get_deepcopy_object_ids(): {self.get_deepcopy_object_ids()}')
            keys_to_delete = self._remove_deleted_objs(keys_for_deleted_objs)
        except Exception as e:
            print(f'Error in recursively_remove_deleted_objs: {repr(e)}')
            traceback.print_exc()
            raise

        # for deleted_key in deleted_keys:
        #     if deleted_key in keys_for_deleted_objs:
        #         keys_for_deleted_objs.remove(deleted_key)
        keys_for_deleted_objs.clear()
        keys_for_deleted_objs.extend(keys_to_delete)

    def _remove_deleted_objs(self, keys_for_deleted_objs: SetDeque[int]) -> SetDeque[int]:
        # keys_to_delete = SetDeque(key for key in keys_for_deleted_objs if key in self)
        keys_to_delete = SetDeque[int](keys_for_deleted_objs)
        # print(f'_remove_deleted_objs({keys_to_delete})')
        # deleted_keys: SetDeque[int] = SetDeque()

        while True:
            any_keys_deleted_this_iteration = False
            retry_keys: SetDeque[int] = SetDeque()
            while len(keys_to_delete) > 0:
                # print(f'len(keys_to_delete): {len(keys_to_delete)}')
                # print(f'keys_to_delete: {keys_to_delete}')
                key = keys_to_delete.popleft()

                if key not in self.data:
                    if key in self._keep_alive_dict:  # happens occasionally with e.g. tuples
                        keys_to_delete = self._add_sub_obj_ids_to_deletion_keys(key, keys_to_delete)
                        self._delete_memo_entry(key)
                    continue

                # obj = self.data[key]
                # print(f'obj: {obj}')
                ref_count = sys.getrefcount(self.data[key])
                # print(f'{key}: {repr(self.data[key])} has {ref_count} references')
                # for k, v in self.data.items():
                #     print(f'{k}: {repr(v)}, id(val)={id(v)}')
                # k = 0
                # v = 0
                ref_count_target = 2
                # print(f'ref_count_target: {ref_count_target}')
                # for i, ref in enumerate(gc.get_referrers(self.data[key])):
                #     try:
                #         print(f'Reference {i}')
                #         print('------------')
                #         print(f'{type(ref)}: {ref}')
                #         print(*gc.get_referrers(ref))
                #     except Exception as e:
                #         print(f'Error: {repr(e)}')
                #         pass
                # del ref
                # loc = locals()
                # print(f'locals(): {loc}')
                # del loc

                if ref_count <= ref_count_target:
                    # print(f'Now deleting {key}: {self.data[key]}')
                    keys_to_delete = self._add_sub_obj_ids_to_deletion_keys(key, keys_to_delete)
                    self._delete_memo_entry(key)
                    any_keys_deleted_this_iteration = True
                    # deleted_keys.append(key)
                else:
                    retry_keys.append(key)

            keys_to_delete = SetDeque(key for key in retry_keys if key in self)
            if not any_keys_deleted_this_iteration:
                break

        return keys_to_delete

    def _add_sub_obj_ids_to_deletion_keys(self, key: int,
                                          keys_to_delete: SetDeque[int]) -> SetDeque[int]:
        if key in self._sub_obj_ids:
            for i, sub_obj_id in enumerate(self._sub_obj_ids[key]):
                if sub_obj_id != key and sub_obj_id not in keys_to_delete:
                    # print(f'Adding {sub_obj_id} to keys_to_delete')
                    keys_to_delete.insert(i, sub_obj_id)

        return keys_to_delete

    def _delete_memo_entry(self, key):
        # print(f'Now deleting {key}: {self[key]}')
        #
        # print(f'memo_dict: {self}')
        # print(f'keep_alive_dict: {self._keep_alive_dict}')
        # print(f'sub_obj_ids: {self._sub_obj_ids}')

        if key in self:
            del self[key]
        if key in self._keep_alive_dict:
            del self._keep_alive_dict[key]
        if key in self._sub_obj_ids:
            del self._sub_obj_ids[key]

__init__ 

__init__() -> None
Source code in src/omnipy/util/memo.py 
41
42
43
44
45
46
def __init__(self) -> None:
    super().__init__()
    self._cur_deepcopy_obj_id: int | None = None
    self._cur_keep_alive_list: list[_ObjT | None] = []
    self._keep_alive_dict: dict[int, _ObjT | None] = {}
    self._sub_obj_ids: defaultdict[int, SetDeque[int]] = defaultdict(SetDeque)

all_are_empty 

all_are_empty(debug: bool = False) -> bool

Check whether all memo and bookkeeping structures are empty.

PARAMETER DESCRIPTION
debug

When True, print a detailed diagnostic report of current internal state and object referrers.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
bool

True when memo data, keep-alive state, and active-copy state are all empty; otherwise False.
Source code in src/omnipy/util/memo.py 
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
def all_are_empty(self, debug: bool = False) -> bool:
    """Check whether all memo and bookkeeping structures are empty.

    Args:
        debug: When ``True``, print a detailed diagnostic report of current
            internal state and object referrers.

    Returns:
        ``True`` when memo data, keep-alive state, and active-copy state are
        all empty; otherwise ``False``.
    """
    _all_are_empty = (
        len(self) == 0 and len(self._keep_alive_dict) == 0 and len(self._sub_obj_ids) == 0
        and self._cur_deepcopy_obj_id is None and len(self._cur_keep_alive_list) == 0)

    if debug:
        print()
        print(f'RefCountMemoDict.all_are_empty(): {_all_are_empty}')
        print('--------------------------------------------------')
        print(f'len(self): {len(self)}')
        print(f'len(self._keep_alive_dict): {len(self._keep_alive_dict)}')
        print(f'len(self._sub_obj_ids): {len(self._sub_obj_ids)}')
        print(f'self._cur_deepcopy_obj_id: {self._cur_deepcopy_obj_id}')
        print(f'len(self._cur_keep_alive_list): {len(self._cur_keep_alive_list)}')

        if len(self) > 0:
            print('==================================================')
            for key in self:
                print(f'Content id: {key} '
                      f'[{sys.getrefcount(self[key]) - 1} refs, id={id(self[key])}]: '
                      f'{repr(self[key])}')

            for key in self:
                print('--------------------------------------------------')
                print(f'References for {key}: {repr(self[key])}')
                print('--------------------------------------------------')
                ref = None
                for i, ref in enumerate(gc.get_referrers(self[key])):
                    try:
                        print(f'[Reference {i}]')
                        print(f'    Type: {type(ref)}')
                        print(f'    ID: {id(ref)}')
                        print(f'    Value: {repr(ref)}')
                        print()

                    except Exception as e:
                        print(f'Error: {repr(e)}')
                        pass
                if ref is not None:
                    del ref

    return _all_are_empty

clear 

clear()

Clear memoized entries and all internal bookkeeping maps.

Source code in src/omnipy/util/memo.py 
101
102
103
104
105
def clear(self):
    """Clear memoized entries and all internal bookkeeping maps."""
    super().clear()
    self._keep_alive_dict.clear()
    self._sub_obj_ids.clear()

get_deepcopy_object_ids 

get_deepcopy_object_ids() -> SetDeque[int]

Return all registered deepcopy root object ids.

RETURNS DESCRIPTION
SetDeque[int]

Object ids currently tracked as deepcopy roots, in insertion order.
Source code in src/omnipy/util/memo.py 
118
119
120
121
122
123
124
def get_deepcopy_object_ids(self) -> SetDeque[int]:
    """Return all registered deepcopy root object ids.

    Returns:
        Object ids currently tracked as deepcopy roots, in insertion order.
    """
    return SetDeque(self._sub_obj_ids.keys())

in_deepcopy_object_ids 

in_deepcopy_object_ids(key: int) -> bool

Check whether an object id is registered as a deepcopy root.

PARAMETER DESCRIPTION
key

Object id to check.

TYPE: int

RETURNS DESCRIPTION
bool

True when the id exists in the deepcopy root registry.
Source code in src/omnipy/util/memo.py 
107
108
109
110
111
112
113
114
115
116
def in_deepcopy_object_ids(self, key: int) -> bool:
    """Check whether an object id is registered as a deepcopy root.

    Args:
        key: Object id to check.

    Returns:
        ``True`` when the id exists in the deepcopy root registry.
    """
    return key in self._sub_obj_ids

keep_alive_after_deepcopy 

keep_alive_after_deepcopy()

Persist objects captured by deepcopy keep-alive bookkeeping.

Non-atomic objects are retained directly in _keep_alive_dict while atomic objects are represented by None placeholders.

Source code in src/omnipy/util/memo.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def keep_alive_after_deepcopy(self):
    """Persist objects captured by deepcopy keep-alive bookkeeping.

    Non-atomic objects are retained directly in ``_keep_alive_dict`` while
    atomic objects are represented by ``None`` placeholders.
    """
    # old_sub_obj_ids = self._sub_obj_ids[self._cur_deepcopy_obj_id]
    # assert len(old_sub_obj_ids) == len(self._cur_keep_alive_list)
    #
    # self._sub_obj_ids[self._cur_deepcopy_obj_id] = SetDeque()

    while len(self._cur_keep_alive_list) > 0:
        keep_alive_obj = self._cur_keep_alive_list.pop()

        # id_keep_alive_obj = id(keep_alive_obj)
        # assert id_keep_alive_obj in old_sub_obj_ids
        # self._sub_obj_ids[self._cur_deepcopy_obj_id].append(id_keep_alive_obj)

        if not self._is_atomic(keep_alive_obj):
            self._keep_alive_dict[id(keep_alive_obj)] = keep_alive_obj
        else:
            self._keep_alive_dict[id(keep_alive_obj)] = None

recursively_remove_deleted_objs 

recursively_remove_deleted_objs(keys_for_deleted_objs: SetDeque[int])

Recursively remove memo entries whose source objects were deleted.

PARAMETER DESCRIPTION
keys_for_deleted_objs

Initial candidate object ids for deletion. The deque is updated in place with ids that could not be removed in the current pass.

TYPE: SetDeque[int]

RAISES DESCRIPTION
Exception

Re-raises unexpected exceptions from internal removal logic after printing debugging details.
Source code in src/omnipy/util/memo.py 
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
def recursively_remove_deleted_objs(
    self,
    keys_for_deleted_objs: SetDeque[int],
):
    """Recursively remove memo entries whose source objects were deleted.

    Args:
        keys_for_deleted_objs: Initial candidate object ids for deletion. The
            deque is updated in place with ids that could not be removed in the
            current pass.

    Raises:
        Exception: Re-raises unexpected exceptions from internal removal logic
            after printing debugging details.
    """
    try:
        # print(f'keys_for_deleted_objs: {keys_for_deleted_objs}')
        # print(f'self.get_deepcopy_object_ids(): {self.get_deepcopy_object_ids()}')
        keys_to_delete = self._remove_deleted_objs(keys_for_deleted_objs)
    except Exception as e:
        print(f'Error in recursively_remove_deleted_objs: {repr(e)}')
        traceback.print_exc()
        raise

    # for deleted_key in deleted_keys:
    #     if deleted_key in keys_for_deleted_objs:
    #         keys_for_deleted_objs.remove(deleted_key)
    keys_for_deleted_objs.clear()
    keys_for_deleted_objs.extend(keys_to_delete)

setup_deepcopy 

setup_deepcopy(obj)

Initialize bookkeeping for a new root-object deepcopy.

PARAMETER DESCRIPTION
obj

Root object about to be deep-copied.
RAISES DESCRIPTION
AssertionError

If a deepcopy is already in progress, temporary keep-alive state is not empty, or this object id is already registered in the memo state.
Source code in src/omnipy/util/memo.py 
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
def setup_deepcopy(self, obj):
    """Initialize bookkeeping for a new root-object deepcopy.

    Args:
        obj: Root object about to be deep-copied.

    Raises:
        AssertionError: If a deepcopy is already in progress, temporary
            keep-alive state is not empty, or this object id is already
            registered in the memo state.
    """
    assert self._cur_deepcopy_obj_id is None, \
        f'self._cur_deepcopy_obj_id is not None, but {self._cur_deepcopy_obj_id}'
    assert len(self._cur_keep_alive_list) == 0, \
        f'len(self._cur_keep_alive_list) == {len(self._cur_keep_alive_list)} instead of 0'

    # Solution to test_ref_count_memo_dict_repeated_deepcopy_same_obj(), but not
    # necessary for the current implementation (see comments on the test). Fix unnecessarily
    # deletes shared fragments between the old and new deepcopies, and is thus disabled.
    # Instead, there is a check to ensure that the current object has not already been
    # deepcopied, which is the assumption behind disabling the fix

    assert id(obj) not in self._sub_obj_ids, \
        (f'Object with id: {id(obj)} already in deepcopy_memo. It has either already been '
         f'deepcopied, or it is reusing the id of a deleted object which has not been deleted '
         'from the deepcopy memo.')

    self._cur_deepcopy_obj_id = id(obj)

teardown_deepcopy 

teardown_deepcopy()

Finalize bookkeeping state after a deepcopy attempt.

This clears active-copy markers and removes root-id tracking if the current root was not preserved in keep-alive storage.

Source code in src/omnipy/util/memo.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def teardown_deepcopy(self):
    """Finalize bookkeeping state after a deepcopy attempt.

    This clears active-copy markers and removes root-id tracking if the
    current root was not preserved in keep-alive storage.
    """
    # Also seems to be unnecessary now. This is for recovering from exceptions during deepcopy
    # that would leave the memo dict in an inconsistent state. This should now be handled by
    # SnapshotHolder.take_snapshot().
    #
    # for possibly_added_obj in self._sub_obj_ids[self._cur_deepcopy_obj_id]:
    #     if possibly_added_obj in self and possibly_added_obj not in self._keep_alive_dict:
    #         del self[possibly_added_obj]

    if self._cur_deepcopy_obj_id and \
            (self._cur_deepcopy_obj_id in self._sub_obj_ids
             and self._cur_deepcopy_obj_id not in self._keep_alive_dict):
        del self._sub_obj_ids[self._cur_deepcopy_obj_id]

    self._cur_deepcopy_obj_id = None
    self._cur_keep_alive_list = []