-
Notifications
You must be signed in to change notification settings - Fork 13
/
lockable.py
351 lines (271 loc) · 10.8 KB
/
lockable.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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
"""
A small mixin to lock attribute and method access at runtime.
Sometimes we wish to restrict users of pyiron from changing certain things past certain stages of object lifetime, e.g.
the input of jobs should only be changed before it is run, but still need to be able to change them internally. This
can be implemented with :class:`~Lockable` and the decorator :func:`~sentinel`. It should be thought of as a well
defined escape hatch that is rarely necessary. Users should never be expected to unlock an object ever again after it
has been locked by them or pyiron.
The context manager functionality is implemented in a separate class rather than directly on Lockable to conserve dunder
name space real estate and let subclasses be context managers on their own.
Through out the code inside methods of `Lockable` will use `object.__setattr__`
and `object.__getattribute__` to avoid any overloading attribute access that
sibling classes may bring in.
"""
from typing import Optional, Literal
from functools import wraps
import warnings
from pyiron_base.interfaces.has_groups import HasGroups
class Locked(Exception):
pass
class LockedWarning(UserWarning):
pass
def sentinel(meth):
"""
Wrap a method to fail if `read_only` is `True` on the owning object.
Use together with :class:`Lockable`.
Args:
meth (method): method to call if `read_only` is `False`.
Returns:
wrapped method
"""
def dispatch_or_error(self, *args, **kwargs):
try:
method = object.__getattribute__(self, "_lock_method")
except AttributeError:
method = None
if method not in ("error", "warning"):
method = "error"
if self.read_only and method == "error":
raise Locked(
"Object is currently locked! Use unlocked() if you know what you are doing."
)
elif self.read_only and method == "warning":
warnings.warn(
f"{meth.__name__} called on {type(self)}, but object is locked!",
category=LockedWarning,
)
return meth(self, *args, **kwargs)
# if sentinel is applied to __setattr__ we must ensure that `read_only`
# stays available, otherwise we can't unlock again later
if meth.__name__ == "__setattr__":
@wraps(meth)
def f(self, *args, **kwargs):
if len(args) > 0:
target = args[0]
else:
target = kwargs["name"]
if target in ("read_only", "_read_only"):
return meth(self, *args, **kwargs)
return dispatch_or_error(self, *args, **kwargs)
else:
f = wraps(meth)(dispatch_or_error)
return f
class _UnlockContext:
"""
Context manager that unlocks and relocks a :class:`Lockable`.
This is an implementation detail of :class:`Lockable`.
"""
__slots__ = ("owner",)
def __init__(self, owner):
self.owner = owner
def __enter__(self):
self.owner.read_only = False
return self.owner
def __exit__(self, *_):
self.owner.lock()
return False # never handle exceptions
def _iterate_lockable_subs(lockable_groups):
"""
Yield sub nodes and groups that are lockable. Recurse into groups.
If the given object is not a :class:`HasGroups` yield nothing.
"""
if not isinstance(lockable_groups, HasGroups):
return
subs = lockable_groups.list_all()
for n in subs["nodes"]:
node = lockable_groups[n]
if isinstance(node, Lockable):
yield node
for g in subs["groups"]:
group = lockable_groups[g]
yield group
yield from _iterate_lockable_subs(g)
class Lockable:
"""
A small mixin to lock attribute and method access at runtime.
The mixin maintains an :attr:`~.read_only` and offers a context manager to temporarily unset it. It does **not**
restrict access to any attributes or methods on its own. Instead sub classes are expected to mark methods they wish
protected with :func:`.sentinel`. Wrapped methods will then raise :exc:`.Locked` if :attr:`~.read_only` is set.
If the subclass also implements :class:`.HasGroups`, locking it will iterate over all nodes and (recursively)
groups and lock them if possible and vice-versa for unlocking.
Once an object has been locked it should generally not be expected to be (permanently) unlocked again, especially
not explicitely by the user.
Subclasses need to initialize this class by calling the inherited `__init__`, if explicitely overriding it. When
not explicitely overriding it (as in the examples below), take care that either the other super classes call
`super().__init__` or place this class before them in the inheritance order. Also be sure to initialize it before
using methods and properties decorated with :func:`.sentinel`.
Subclasses may override :meth:`_on_lock` and :meth:`_on_unlock` if they wish to customize locking/unlocking
behaviour, provided that they call `super()` in their overloads.
Let's start with a simple example; a list that can be locked
>>> class LockList(Lockable, list):
... __setitem__ = sentinel(list.__setitem__)
... clear = sentinel(list.clear)
>>> l = LockList([1,2,3])
>>> l
[1, 2, 3]
Deriving adds the read only flag
>>> l.read_only
False
While it is not set, we may mutate the object
>>> l[2] = 4
>>> l[2]
4
Once it is locked, the wrapped methods will raise errors
>>> l.lock()
>>> l.read_only
True
>>> l[1]
2
>>> l[1] = 4
Traceback (most recent call last):
...
lockable.Locked: Object is currently locked! Use unlocked() if you know what you are doing.
You can lock an object multiple times to no effect
>>> l.lock()
From now on every modification should be done with the :meth:`~.unlocked()` context manager. It returns the
unlocked object itself.
>>> with l.unlocked():
... l[1] = 4
>>> l[1]
4
>>> with l.unlocked() as lopen:
... print(l is lopen)
... l[1] = 4
True
:func:`~.sentinel` can be used for methods, item and attribute access.
>>> l.clear()
Traceback (most recent call last):
...
lockable.Locked: Object is currently locked! Use unlocked() if you know what you are doing.
>>> with l.unlocked():
... l.clear()
>>> l
[]
When used together with :class:`.HasGroups`, objects will be locked recursively.
>>> class LockGroupDict(Lockable, dict, HasGroups):
... __setitem__ = sentinel(dict.__setitem__)
...
... def _list_groups(self):
... return [k for k, v in self.items() if isinstance(v, LockGroupDict)]
...
... def _list_nodes(self):
... return [k for k, v in self.items() if not isinstance(v, LockGroupDict)]
>>> d = LockGroupDict(a=dict(c=1, d=2), b=LockGroupDict(c=1, d=2))
>>> d.lock()
Since the first item is a plain dict, it can still be mutated.
>>> type(d['a'])
<class 'dict'>
>>> d['a']['c'] = 23
>>> d['a']['c']
23
Where as the second will be locked from now on
>>> type(d['b'])
<class 'lockable.LockGroupDict'>
>>> d['b']['c'] = 23
Traceback (most recent call last):
...
lockable.Locked: Object is currently locked! Use unlocked() if you know what you are doing.
>>> d['b']['c']
1
but we can unlock it as usual
>>> with d.unlocked():
... d['b']['d'] = 23
>>> d['b']['d']
23
To use this class with properties, simply decorate the setter
>>> class MyLock(Lockable):
... def __init__(self, foo):
... super().__init__()
... self._foo = foo
... @property
... def foo(self):
... return self._foo
... @foo.setter
... @sentinel
... def foo(self, value):
... self._foo = value
>>> ml = MyLock(42)
>>> ml.foo
42
>>> ml.foo = 23
>>> ml.lock()
>>> ml.foo = 42
Traceback (most recent call last):
...
lockable.Locked: Object is currently locked! Use unlocked() if you know what you are doing.
It's possible to change the errors raised into a warning and allow
modification by passing `lock_method` to :meth:`~.Lockable.__init__` or
`method` to :meth:`~.lock`.
>>> mw = LockList(lock_method="warning")
>>> mw.append(0)
>>> mw.lock()
>>> mw[0] = 1 # will print the warning
>>> mw[0]
1
>>> mw = LockList()
>>> mw.append(0)
>>> mw.lock(method='warning')
>>> mw[0] = 1 # will print the warning
>>> mw[0]
1
"""
def __init__(self, *args, lock_method: str = "error", **kwargs):
object.__setattr__(self, "_read_only", False)
object.__setattr__(self, "_lock_method", lock_method)
super().__init__(*args, **kwargs)
@property
def read_only(self) -> bool:
"""
bool: False if the object can currently be written to
Setting this value will trigger :meth:`._on_lock` and :meth:`._on_unlock` if it changes.
"""
return object.__getattribute__(self, "_read_only")
@read_only.setter
def read_only(self, value: bool):
changed = self._read_only != value
if changed:
self._read_only = value
if value:
self._on_lock()
else:
self._on_unlock()
def _on_lock(self):
for it in _iterate_lockable_subs(self):
it.lock()
def _on_unlock(self):
for it in _iterate_lockable_subs(self):
it.read_only = False
def lock(self, method: Optional[Literal["error", "warning"]] = None):
"""
Set :attr:`~.read_only`.
Objects may be safely locked multiple times without further effect.
Args:
method (str, either "error" or "warning"): if "error" raise an :class:`.Locked` exception if modification is
attempted; if "warning" raise a :class:`.LockedWarning` warning; default is "error" or the value
passed to the constructor.
Raises:
ValueError: if `method` is not an allowed value
"""
if method not in ["error", "warning", None]:
raise ValueError(f"Unrecognized lock method {method}!")
if method is not None:
object.__setattr__(self, "_lock_method", method)
self.read_only = True
def unlocked(self) -> _UnlockContext:
"""
Unlock the object temporarily.
Context manager returns this object again and relocks it after the `with` statement finished.
.. note:: `lock()` vs. `unlocked()`
There is a small asymmetry between these two methods. :meth:`.lock` can only be done once (meaningfully), while :meth:`.unlocked` is a context manager and can be called multiple times.
"""
return _UnlockContext(self)