A lockable/unlockable dictionary. After locking, any in-place modifications will raise TypeError. By default auto_wrap=True, which recursively converts nested dict objects in dict/list/tuple/set to LockableDict, so that recursive locking works consistently both internally and externally
| 3 | |
| 4 | |
| 5 | class LockableDict(dict): |
| 6 | """ |
| 7 | A lockable/unlockable dictionary. After locking, any in-place modifications will raise TypeError. |
| 8 | By default auto_wrap=True, which recursively converts nested dict objects in dict/list/tuple/set |
| 9 | to LockableDict, so that recursive locking works consistently both internally and externally. |
| 10 | """ |
| 11 | |
| 12 | def __init__(self, *args, auto_wrap: bool = True, **kwargs): |
| 13 | self._locked: bool = False |
| 14 | self._auto_wrap: bool = auto_wrap |
| 15 | # Build with temporary dict, then wrap uniformly before writing to self, avoiding bypass of __setitem__ |
| 16 | tmp = dict(*args, **kwargs) |
| 17 | for k, v in tmp.items(): |
| 18 | dict.__setitem__(self, k, self._wrap(v)) |
| 19 | |
| 20 | # ========== Public API ========== |
| 21 | @property |
| 22 | def locked(self) -> bool: |
| 23 | return self._locked |
| 24 | |
| 25 | def lock(self, recursive: bool = True) -> None: |
| 26 | """Lock the dictionary. When recursive=True, also recursively locks nested LockableDict objects.""" |
| 27 | self._locked = True |
| 28 | if recursive: |
| 29 | for v in self.values(): |
| 30 | if isinstance(v, LockableDict): |
| 31 | v.lock(True) |
| 32 | |
| 33 | def unlock(self, recursive: bool = True) -> None: |
| 34 | """Unlock the dictionary. When recursive=True, also recursively unlocks nested LockableDict objects.""" |
| 35 | self._locked = False |
| 36 | if recursive: |
| 37 | for v in self.values(): |
| 38 | if isinstance(v, LockableDict): |
| 39 | v.unlock(True) |
| 40 | |
| 41 | @contextmanager |
| 42 | def temporarily_unlocked(self, recursive: bool = True): |
| 43 | """ |
| 44 | Temporarily unlock in context manager form, restoring original state on exit. |
| 45 | Typical usage: |
| 46 | with d.temporarily_unlocked(): |
| 47 | d["x"] = 1 |
| 48 | """ |
| 49 | prev = self._locked |
| 50 | if prev and recursive: |
| 51 | # First temporarily unlock all child nodes as well |
| 52 | stack: list[LockableDict] = [] |
| 53 | |
| 54 | def _collect(node: "LockableDict"): |
| 55 | for v in node.values(): |
| 56 | if isinstance(v, LockableDict): |
| 57 | stack.append(v) |
| 58 | _collect(v) |
| 59 | |
| 60 | _collect(self) |
| 61 | self._locked = False |
| 62 | for n in stack: |
no outgoing calls