Skip to content

omnipy.data.snapshot

CLASS DESCRIPTION
SnapshotHolder
SnapshotWrapper
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]

METHOD DESCRIPTION
__init__
all_are_empty
clear
delete_scheduled_deepcopy_content_ids
get
get_deepcopy_content_ids
get_deepcopy_content_ids_scheduled_for_deletion
in_deepcopy_content_ids
schedule_deepcopy_content_ids_for_deletion
take_snapshot
take_snapshot_setup
take_snapshot_teardown
Source code in src/omnipy/data/snapshot.py 
 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
class SnapshotHolder(WeakKeyRefContainer[HasContentT, IsSnapshotWrapper[HasContentT, ContentT]],
                     Generic[HasContentT, ContentT]):
    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:
        self._deepcopy_content_ids_for_deleted_objs.clear()
        self._deepcopy_memo.clear()
        super().clear()

    def all_are_empty(self, debug: bool = False) -> bool:
        _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 self._deepcopy_memo.in_deepcopy_object_ids(key)

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

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

    def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
        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:
        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:
        gc.disable()

    def take_snapshot_teardown(self) -> None:
        self.delete_scheduled_deepcopy_content_ids()
        gc.enable()

    def take_snapshot(self, obj: HasContentT) -> None:
        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 
33
34
35
36
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
Source code in src/omnipy/data/snapshot.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def all_are_empty(self, debug: bool = False) -> bool:
    _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
Source code in src/omnipy/data/snapshot.py 
42
43
44
45
def clear(self) -> None:
    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
Source code in src/omnipy/data/snapshot.py 
86
87
88
89
90
91
def delete_scheduled_deepcopy_content_ids(self) -> None:
    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
Source code in src/omnipy/util/weak.py 
26
27
28
29
30
31
def get(self, key: _AnyKeyT) -> _ValT | None:
    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]
Source code in src/omnipy/data/snapshot.py 
75
76
def get_deepcopy_content_ids(self) -> SetDeque[int]:
    return self._deepcopy_memo.get_deepcopy_object_ids()

get_deepcopy_content_ids_scheduled_for_deletion 

get_deepcopy_content_ids_scheduled_for_deletion() -> SetDeque[int]
Source code in src/omnipy/data/snapshot.py 
78
79
def get_deepcopy_content_ids_scheduled_for_deletion(self) -> SetDeque[int]:
    return self._deepcopy_content_ids_for_deleted_objs

in_deepcopy_content_ids 

in_deepcopy_content_ids(key: int) -> bool
Source code in src/omnipy/data/snapshot.py 
72
73
def in_deepcopy_content_ids(self, key: int) -> bool:
    return self._deepcopy_memo.in_deepcopy_object_ids(key)

schedule_deepcopy_content_ids_for_deletion 

schedule_deepcopy_content_ids_for_deletion(*keys: int) -> None
Source code in src/omnipy/data/snapshot.py 
81
82
83
84
def schedule_deepcopy_content_ids_for_deletion(self, *keys: int) -> None:
    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
Source code in src/omnipy/data/snapshot.py 
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
def take_snapshot(self, obj: HasContentT) -> None:
    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
Source code in src/omnipy/data/snapshot.py 
93
94
def take_snapshot_setup(self) -> None:
    gc.disable()

take_snapshot_teardown 

take_snapshot_teardown() -> None
Source code in src/omnipy/data/snapshot.py 
96
97
98
def take_snapshot_teardown(self) -> None:
    self.delete_scheduled_deepcopy_content_ids()
    gc.enable()

SnapshotWrapper dataclass

Bases: Generic[ObjContraT, ContentT]

METHOD DESCRIPTION
__init__
differs_from
taken_of_same_obj
ATTRIBUTE DESCRIPTION
id

TYPE: int

snapshot

TYPE: ContentT

Source code in src/omnipy/data/snapshot.py 
15
16
17
18
19
20
21
22
23
24
@dataclass
class SnapshotWrapper(Generic[ObjContraT, ContentT]):
    id: int
    snapshot: ContentT

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

    def differs_from(self, obj: ObjContraT) -> bool:
        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
Source code in src/omnipy/data/snapshot.py 
23
24
def differs_from(self, obj: ObjContraT) -> bool:
    return not all_equals(self.snapshot, obj)

taken_of_same_obj 

taken_of_same_obj(obj: ObjContraT) -> bool
Source code in src/omnipy/data/snapshot.py 
20
21
def taken_of_same_obj(self, obj: ObjContraT) -> bool:
    return self.id == id(obj)