Coverage for src / typed_event.py: 100%

1""" 

2``@event`` decorator for methods, to make them into subscribable events. 

3 

4.. default-role:: py:obj 

5""" 

6 

7__all__ = [ 

8 "event", 

9 "Event", 

10 "CancelEvent", 

11] 

12 

13import logging 

14import traceback 

15import inspect 

16from functools import update_wrapper 

17from typing import ( 

18 Callable, 

19 Literal, 

20 ParamSpec, 

21 Generic, 

22 Self, 

23 TypeAlias, 

24 TypeVar, 

25 overload, 

26 get_args, 

27) 

28 

29from weakref import WeakValueDictionary 

30 

31# For keeping bound copies. I also tried to make listeners weak-referenced. 

32# Turns out that weak-referencing listeners is not that great after all, because 

33# it breaks using lambda or inner functions as handlers. They are immediately 

34# lost when the defining scope exits. 

35 

36T = TypeVar("T") 

37P = ParamSpec("P") 

38R = TypeVar("R") # return type 

39 

40ExceptionPolicy: TypeAlias = Literal["log", "print", "raise", "group"] 

41 

42 

43class CancelEvent(Exception): 

44 """Raise this in an event handler to inhibit all further processing.""" 

45 

46 

47class Event(Generic[P, R]): 

48 """Notifies a number of "listeners" (functions) when called. 

49 

50 The principle is well-known under many names: 

51 

52 * Define the event as member of a class that wants to tell the world 

53 about changes. 

54 * Arbitrary listeners can subscribe the event. 

55 * In the class's implementation, the event is called when the trigger 

56 condition occurs. Listeners will be called in the order they subscribed. 

57 

58 **Defining events** 

59 

60 In contrast to other adhoc event systems, this one enforces well-defined 

61 signatures and documentation. An event is created by decorating a method 

62 (the so-called "prototype") with ``@event``. 

63 

64 The ``event`` will take over the method's signature, annotations and 

65 docstring. IDE tools and Sphinx documentation should (mostly) "see" the 

66 Event like any other method. 

67 

68 The prototype method is executed every time the event is triggered. Usually 

69 it does not need any code except for a docstring or ``pass`` statement. 

70 

71 Restrictions apply: 

72 

73 * In ``strict`` mode, only positional-only and/or keyword-only 

74 args are allowed. This is to make clear to the user how the arguments 

75 will be given (by position or by name). 

76 

77 * Yes: ``prototype(a: int, b: str, /)`` 

78 * Yes: ``prototype(*, a: int, b: str)`` 

79 * No: ``prototype(a:int, b:str)`` (but allowed in non-strict mode) 

80 

81 ``strict`` mode is disabled for backwards compatibility, but will become 

82 the default in the future. 

83 

84 * Arguments with default values are forbidden, since their meaning would be 

85 ambiguous for the user of the class. 

86 * The prototype does not get an automatic ``self`` argument. I.e. it works 

87 like a ``staticmethod``. You *can* define a ``self`` argument, but it must 

88 be given explicitly upon calling. 

89 

90 **Listeners** 

91 

92 A listener is a Callable whose signature fits the event specification. 

93 Event listeners can be subscribed/unsubscribed using the ``+=`` and ``-=`` 

94 operators. Listener signature is *not* checked at the time of subscription. 

95 

96 There is some freedom in listener signature. E.g. you can have extra 

97 parameters with default values, or you can catch the event data via 

98 ``*args`` / ``**kwargs``. 

99 

100 **Triggering the event** 

101 

102 The event is triggered by calling the ``event`` instance. Usually this 

103 happens within the class containing the Event. 

104 

105 First, the wrapped protoype is executed, in order to verify correct arguments. 

106 Note that adherence to annotated types is *not* checked, in line with 

107 standard Python behavior. 

108 

109 Any handler can raise `CancelEvent` to gracefully abort the processing of 

110 further listeners. 

111 

112 **Return values** 

113 

114 At most one listener is expected to return a value. If multiple listeners 

115 return a value, an exception is raised. 

116 

117 The return type must always be Optional. 

118 

119 **Exceptions** 

120 

121 Listeners may raise exceptions that are unexpected for the event's origin 

122 site. `Event` has the "exceptions" parameter to control how they are 

123 handled: 

124 

125 * ``"log"`` (default) emits a ``logging.error`` message with the traceback. 

126 * ``"print"`` prints the exception (using ``traceback.print_exception``). 

127 * ``"raise"`` raises any exception immediately. No subsequent listeners are 

128 called. 

129 * ``"group"`` calls all listeners, then raises an ``ExceptionGroup`` if any 

130 failed. The error is always an ``ExceptionGroup``, even in case of a single 

131 error. 

132 

133 When using ``raise``, code that triggers an event must be prepared for any 

134 exception being thrown at it. 

135 

136 `CancelEvent` is obviously exempt from this exception handling. 

137 

138 **Unbound/Bound distinction** 

139 

140 Analogous to unbound methods, the class will contain the event as "unbound" 

141 event. You can in principle subscribe to it, and trigger it using 

142 ``Class.event()``. There is only one, global list of subscribers. 

143 

144 A class *instance* will have a "bound" copy of the ``Event``, meaning that 

145 it has its own list of subscribers independent from all other instances. It 

146 does *not* inherit listeners from the unbound event. Typically, the *bound* 

147 event is the one you want to subscribe to. 

148 

149 Lastly, you can also apply ``@event`` to a module-level function. There will 

150 be only one, global list of subscribers, same as for an unbound event. 

151 

152 **Example**:: 

153 

154 # Class definition 

155 class MyCounter: 

156 @event 

157 def counter_changed_to(self, new_value:int): 

158 '''Event: counter changed to given value''' 

159 

160 def my_timer_function(self): 

161 # ... 

162 self.counter_changed(123) 

163 # ... 

164 

165 # User code 

166 class MyGUI: 

167 def __init__(self, counter_instance:MyCounter): 

168 self.counter_instance = counter_instance 

169 self.counter_instance.counter_changed_to += self.on_counter_changed 

170 

171 def on_counter_changed(self, new_value): 

172 self.update_display(new_value) 

173 """ 

174 

175 def __init__( 

176 self, 

177 prototype: Callable[P, R], 

178 strict: bool | None = None, 

179 exceptions: ExceptionPolicy = "log", 

180 ): 

181 self._prototype = prototype 

182 self._listeners: list[Callable[P, R | None]] = [] 

183 # None as default, so that we can discern from excplicit opt-in. 

184 # allows to add a warning in the future. 

185 self._strict: bool = strict or False 

186 if exceptions not in (policies := get_args(ExceptionPolicy)): 

187 raise ValueError(f"exceptions must be one of {policies}") 

188 self._exceptions = exceptions 

189 

190 sig: inspect.Signature = inspect.signature(self._prototype) 

191 iP = inspect.Parameter 

192 if any(p.default is not iP.empty for p in sig.parameters.values()): 

193 raise TypeError("Default values are forbidden for events") 

194 if any( 

195 p.kind in (iP.VAR_POSITIONAL, iP.VAR_KEYWORD) 

196 for p in sig.parameters.values() 

197 ): 

198 raise TypeError("*args and **kwargs are forbidden for events") 

199 if self._strict and any( 

200 p.kind == iP.POSITIONAL_OR_KEYWORD for p in sig.parameters.values() 

201 ): 

202 raise TypeError( 

203 "Event arguments must be marked positional-only or keyword-only!" 

204 ) 

205 self._self_arg = "self" in sig.parameters 

206 self._argnames = [p.name for p in sig.parameters.values() if p.name != "self"] 

207 update_wrapper(self, self._prototype) 

208 

209 self._is_bound = False 

210 self._bound_copies = WeakValueDictionary() 

211 

212 def __get__(self, instance, owner) -> "Event[P, R]": 

213 # Copy the event for each instance, so that that each instance 

214 # has its private list of listeners. 

215 if instance is None: 

216 return self 

217 key = id(instance) 

218 try: 

219 return self._bound_copies[key] 

220 except KeyError: 

221 ev = Event(self._prototype, strict=self._strict) 

222 ev._is_bound = True 

223 self._bound_copies[key] = ev 

224 return ev 

225 

226 def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None: 

227 epolicy = self._exceptions 

228 results = [] 

229 # Call to verify arguments 

230 r = self._prototype(*args, **kwargs) 

231 if r is not None: 

232 raise RuntimeError("Event prototype should not return a value") 

233 # === Call each listener === 

234 excs = [] 

235 for listener in self._listeners: 

236 try: 

237 r = listener(*args, **kwargs) 

238 except CancelEvent: 

239 break 

240 except Exception as exc: 

241 if epolicy == "log": 

242 logging.exception(exc) 

243 elif epolicy == "print": 

244 traceback.print_exception(exc) 

245 elif epolicy == "group": 

246 excs.append(exc) 

247 else: 

248 raise 

249 if r is not None: 

250 results.append(r) 

251 if excs: 

252 raise ExceptionGroup("One or more listeners raised an error.", excs) 

253 if len(results) > 1: 

254 raise RuntimeError("Multple return values from event handler") 

255 return None if not results else results[0] 

256 

257 def __iadd__(self, listener: Callable[P, R | None]) -> Self: 

258 # Old handlers are most likely to vanish when new ones are added :-) 

259 self._listeners.append(listener) 

260 return self 

261 

262 def __isub__(self, listener: Callable[P, R | None]) -> Self: 

263 self._listeners = [ 

264 r_listener for r_listener in self._listeners if r_listener is not listener 

265 ] 

266 return self 

267 

268 def __str__(self): 

269 names = ", ".join(self._argnames) 

270 prefix = "Bound" if self._is_bound else "Unbound" 

271 return f"<{prefix} Event {self._prototype.__qualname__}({names})>" 

272 

273 __repr__ = __str__ 

274 

275 

276# Decorator: Allow both @event and @event(params=...) syntax. 

277 

278 

279# Used as @event without parens 

280@overload 

281def event( 

282 prototype: Callable[P, R], 

283 *, 

284 strict: bool | None = None, 

285 exceptions: ExceptionPolicy = "log", 

286) -> Event[P, R]: ... 

287 

288 

289# Used as @event(...) 

290@overload 

291def event( 

292 prototype: None = None, 

293 *, 

294 strict: bool | None = None, 

295 exceptions: ExceptionPolicy = "log", 

296) -> Callable[[Callable[P, R]], Event[P, R]]: ... 

297 

298 

299def event( 

300 prototype: Callable[P, R] | None = None, 

301 *, 

302 strict: bool | None = None, 

303 exceptions: ExceptionPolicy = "log", 

304) -> Event[P, R] | Callable[[Callable[P, R]], Event[P, R]]: 

305 """Turn the decorated method into an Event. 

306 

307 See `Event`. The `@event` decorator allows to pass arguments: 

308 

309 @event(strict=False, exeptions="print") 

310 def some_event(arg1: bool, /): ... 

311 """ 

312 if prototype is None: 

313 

314 def wrap(prototype): 

315 return Event(prototype, strict=strict, exceptions=exceptions) 

316 

317 return wrap 

318 else: 

319 return Event(prototype, strict=strict, exceptions=exceptions)