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
|
# --------------------------------------------------------------------------------------
# Copyright (c) 2023-2024, Nucleic Development Team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# --------------------------------------------------------------------------------------
"""Tools to declare static observers in Atom subclasses"""
from types import FunctionType
from typing import (
TYPE_CHECKING,
Any,
Callable,
List,
Mapping,
Optional,
Tuple,
TypeVar,
Union,
)
from ..catom import ChangeType
from ..typing_utils import ChangeDict
if TYPE_CHECKING:
from ..atom import Atom
def observe(*names: str, change_types: ChangeType = ChangeType.ANY) -> "ObserveHandler":
"""A decorator which can be used to observe members on a class.
Parameters
----------
*names
The str names of the attributes to observe on the object.
These must be of the form 'foo' or 'foo.bar'.
change_types
The flag specifying the type of changes to observe.
"""
# backwards compatibility for a single tuple or list argument
if len(names) == 1 and isinstance(names[0], (tuple, list)):
names = names[0]
pairs: List[Tuple[str, Optional[str]]] = []
for name in names:
if not isinstance(name, str):
msg = "observe attribute name must be a string, got '%s' instead"
raise TypeError(msg % type(name).__name__)
ndots = name.count(".")
if ndots > 1:
msg = "cannot observe '%s', only a single extension is allowed"
raise TypeError(msg % name)
if ndots == 1:
name, attr = name.split(".")
pairs.append((name, attr))
else:
pairs.append((name, None))
return ObserveHandler(pairs, change_types)
T = TypeVar("T", bound="Atom")
class ObserveHandler(object):
"""An object used to temporarily store observe decorator state."""
__slots__ = ("pairs", "func", "funcname", "change_types")
#: List of 2-tuples which stores the pair information for the observers.
pairs: List[Tuple[str, Optional[str]]]
#: Callable to be used as observer callback.
func: Optional[Callable[[Mapping[str, Any]], None]]
#: Name of the callable. Used by the metaclass.
funcname: Optional[str]
#: Types of changes to listen to.
change_types: ChangeType
def __init__(
self,
pairs: List[Tuple[str, Optional[str]]],
change_types: ChangeType = ChangeType.ANY,
) -> None:
"""Initialize an ObserveHandler.
Parameters
----------
pairs : list
The list of 2-tuples which stores the pair information for the observers.
"""
self.pairs = pairs
self.change_types = change_types
self.func = None # set by the __call__ method
self.funcname = None
def __call__(
self,
func: Union[
Callable[[ChangeDict], None],
Callable[[T, ChangeDict], None],
# AtomMeta will replace ObserveHandler in the body of an atom
# class allowing to access it for example in a subclass. We lie here by
# giving ObserverHandler.__call__ a signature compatible with an
# observer to mimic this behavior.
ChangeDict,
],
) -> "ObserveHandler":
"""Called to decorate the function.
Parameters
----------
func
Should be either a callable taking as single argument the change
dictionary or a method declared on an Atom object.
"""
assert isinstance(func, FunctionType), "func must be a function"
self.func = func
return self
def clone(self) -> "ObserveHandler":
"""Create a clone of the sentinel."""
clone = type(self)(self.pairs, self.change_types)
clone.func = self.func
return clone
class ExtendedObserver(object):
"""A callable object used to implement extended observers."""
__slots__ = ("funcname", "attr")
#: Name of the function on the owner object which should be used as the observer.
funcname: str
#: Attribute name on the target object which should be observed.
attr: str
def __init__(self, funcname: str, attr: str) -> None:
"""Initialize an ExtendedObserver.
Parameters
----------
funcname : str
The function name on the owner object which should be
used as the observer.
attr : str
The attribute name on the target object which should be
observed.
"""
self.funcname = funcname
self.attr = attr
def __call__(self, change: ChangeDict) -> None:
"""Handle a change of the target object.
This handler will remove the old observer and attach a new
observer to the target attribute. If the target object is not
an Atom object, an exception will be raised.
"""
from ..atom import Atom
old = None
new = None
ctype = change["type"]
if ctype == "create":
new = change["value"]
elif ctype == "update":
old = change["oldvalue"]
new = change["value"]
elif ctype == "delete":
old = change["value"]
attr = self.attr
owner = change["object"]
handler = getattr(owner, self.funcname)
if isinstance(old, Atom):
old.unobserve(attr, handler)
if isinstance(new, Atom):
new.observe(attr, handler)
elif new is not None:
msg = "cannot attach observer '%s' to non-Atom %s"
raise TypeError(msg % (attr, new))
|
