Coverage for src/paperap/signals.py: 97%
140 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-22 16:02 -0400
1"""
5----------------------------------------------------------------------------
7METADATA:
9File: signals.py
10 Project: paperap
11Created: 2025-03-09
12 Version: 0.0.9
13Author: Jess Mann
14Email: jess@jmann.me
15 Copyright (c) 2025 Jess Mann
17----------------------------------------------------------------------------
19LAST MODIFIED:
212025-03-09 By Jess Mann
23"""
25from __future__ import annotations
27import logging
28from collections import defaultdict
29from typing import (
30 Any,
31 Callable,
32 Generic,
33 Literal,
34 Self,
35 TypeAlias,
36 TypedDict,
37 TypeVar,
38 final,
39 overload,
40)
42logger = logging.getLogger(__name__)
45class QueueType(TypedDict):
46 """
47 A type used by SignalRegistry for storing queued signal actions.
48 """
50 connect: dict[str, set[tuple[Callable[..., Any], int]]]
51 disconnect: dict[str, set[Callable[..., Any]]]
52 disable: dict[str, set[Callable[..., Any]]]
53 enable: dict[str, set[Callable[..., Any]]]
56ActionType = Literal["connect", "disconnect", "disable", "enable"]
59@final
60class SignalPriority:
61 """
62 Priority levels for signal handlers.
64 Any int can be provided, but these are the recommended values.
65 """
67 FIRST = 0
68 HIGH = 25
69 NORMAL = 50
70 LOW = 75
71 LAST = 100
74class SignalParams(TypedDict):
75 """
76 A type used by SignalRegistry for storing signal parameters.
77 """
79 name: str
80 description: str
83class Signal[_ReturnType]:
84 """
85 A signal that can be connected to and emitted.
87 Handlers can be registered with a priority to control execution order.
88 Each handler receives the output of the previous handler as its first argument,
89 enabling a filter/transformation chain.
90 """
92 name: str
93 description: str
94 _handlers: dict[int, list[Callable[..., _ReturnType]]]
95 _disabled_handlers: set[Callable[..., _ReturnType]]
97 def __init__(self, name: str, description: str = "") -> None:
98 self.name = name
99 self.description = description
100 self._handlers = defaultdict(list)
101 self._disabled_handlers = set()
102 super().__init__()
104 def connect(self, handler: Callable[..., _ReturnType], priority: int = SignalPriority.NORMAL) -> None:
105 """
106 Connect a handler to this signal.
108 Args:
109 handler: The handler function to be called when the signal is emitted.
110 priority: The priority level for this handler (lower numbers execute first).
112 """
113 self._handlers[priority].append(handler)
115 # Check if the handler was temporarily disabled in the registry
116 if SignalRegistry.get_instance().is_queued("disable", self.name, handler):
117 self._disabled_handlers.add(handler)
119 def disconnect(self, handler: Callable[..., _ReturnType]) -> None:
120 """
121 Disconnect a handler from this signal.
123 Args:
124 handler: The handler to disconnect.
126 """
127 for priority in self._handlers:
128 if handler in self._handlers[priority]:
129 self._handlers[priority].remove(handler)
131 @overload
132 def emit(self, value: _ReturnType | None, *args: Any, **kwargs: Any) -> _ReturnType | None: ...
134 @overload
135 def emit(self, **kwargs: Any) -> _ReturnType | None: ...
137 def emit(self, *args: Any, **kwargs: Any) -> _ReturnType | None:
138 """
139 Emit the signal, calling all connected handlers in priority order.
141 Each handler receives the output of the previous handler as its first argument.
142 Other arguments are passed unchanged.
144 Args:
145 *args: Positional arguments to pass to handlers.
146 **kwargs: Keyword arguments to pass to handlers.
148 Returns:
149 The final result after all handlers have processed the data.
151 """
152 current_value: _ReturnType | None = None
153 remaining_args = args
154 if args:
155 # Start with the first argument as the initial value
156 current_value = args[0]
157 remaining_args = args[1:]
159 # Get all priorities in ascending order (lower numbers execute first)
160 priorities = sorted(self._handlers.keys())
162 # Process handlers in priority order
163 for priority in priorities:
164 for handler in self._handlers[priority]:
165 if handler not in self._disabled_handlers:
166 # Pass the current value as the first argument, along with any other args
167 current_value = handler(current_value, *remaining_args, **kwargs)
169 return current_value
171 def disable(self, handler: Callable[..., _ReturnType]) -> None:
172 """
173 Temporarily disable a handler without disconnecting it.
175 Args:
176 handler: The handler to disable.
178 """
179 self._disabled_handlers.add(handler)
181 def enable(self, handler: Callable[..., _ReturnType]) -> None:
182 """
183 Re-enable a temporarily disabled handler.
185 Args:
186 handler: The handler to enable.
188 """
189 if handler in self._disabled_handlers:
190 self._disabled_handlers.remove(handler)
193class SignalRegistry:
194 """
195 Registry of all signals in the application.
197 Signals can be created, connected to, and emitted through the registry.
199 Examples:
200 >>> SignalRegistry.emit(
201 ... "document.save:success",
202 ... "Fired when a document has been saved successfully",
203 ... kwargs = {"document": document}
204 ... )
206 >>> filtered_data = SignalRegistry.emit(
207 ... "document.save:before",
208 ... "Fired before a document is saved. Optionally filters the data that will be saved.",
209 ... args = (data,),
210 ... kwargs = {"document": document}
211 ... )
213 >>> SignalRegistry.connect("document.save:success", my_handler)
215 """
217 _instance: Self
218 _signals: dict[str, Signal[Any]]
219 _queue: QueueType
221 def __init__(self) -> None:
222 self._signals = {}
223 self._queue = {
224 "connect": {}, # {signal_name: {(handler, priority), ...}}
225 "disconnect": {}, # {signal_name: {handler, ...}}
226 "disable": {}, # {signal_name: {handler, ...}}
227 "enable": {}, # {signal_name: {handler, ...}}
228 }
229 super().__init__()
231 def __new__(cls) -> Self:
232 """
233 Ensure that only one instance of the class is created.
235 Returns:
236 The singleton instance of this class.
238 """
239 if not hasattr(cls, "_instance"):
240 cls._instance = super().__new__(cls)
241 return cls._instance
243 @classmethod
244 def get_instance(cls) -> Self:
245 """
246 Get the singleton instance of this class.
248 Returns:
249 The singleton instance of this class.
251 """
252 if not hasattr(cls, "_instance"):
253 cls._instance = cls()
254 return cls._instance # type: ignore # mypy issue with Self return type
256 def register(self, signal: Signal[Any]) -> None:
257 """
258 Register a signal and process queued actions.
260 Args:
261 signal: The signal to register.
263 """
264 self._signals[signal.name] = signal
266 # Process queued connections
267 for handler, priority in self._queue["connect"].pop(signal.name, set()):
268 signal.connect(handler, priority)
270 # Process queued disconnections
271 for handler in self._queue["disconnect"].pop(signal.name, set()):
272 signal.disconnect(handler)
274 # Process queued disables
275 for handler in self._queue["disable"].pop(signal.name, set()):
276 signal.disable(handler)
278 # Process queued enables
279 for handler in self._queue["enable"].pop(signal.name, set()):
280 signal.enable(handler)
282 def queue_action(self, action: ActionType, name: str, handler: Callable[..., Any], priority: int | None = None) -> None:
283 """
284 Queue any signal-related action to be processed when the signal is registered.
286 Args:
287 action: The action to queue (connect, disconnect, disable, enable).
288 name: The signal name.
289 handler: The handler function to queue.
290 priority: The priority level for this handler (only for connect action).
292 Raises:
293 ValueError: If the action is invalid.
295 """
296 if action not in self._queue:
297 raise ValueError(f"Invalid queue action: {action}")
299 if action == "connect":
300 # If it's in the disconnect queue, remove it
301 priority = priority if priority is not None else SignalPriority.NORMAL
302 self._queue[action].setdefault(name, set()).add((handler, priority))
303 else:
304 # For non-connect actions, just add the handler without priority
305 self._queue[action].setdefault(name, set()).add(handler)
307 def get(self, name: str) -> Signal[Any] | None:
308 """
309 Get a signal by name.
311 Args:
312 name: The signal name.
314 Returns:
315 The signal instance, or None if not found.
317 """
318 return self._signals.get(name)
320 def list_signals(self) -> list[str]:
321 """
322 List all registered signal names.
324 Returns:
325 A list of signal names.
327 """
328 return list(self._signals.keys())
330 def create[R](self, name: str, description: str = "", return_type: type[R] | None = None) -> Signal[R]:
331 """
332 Create and register a new signal.
334 Args:
335 name: Signal name
336 description: Optional description for new signals
337 return_type: Optional return type for new signals
339 Returns:
340 The new signal instance.
342 """
343 signal = Signal[R](name, description)
344 self.register(signal)
345 return signal
347 @overload
348 def emit[_ReturnType](
349 self,
350 name: str,
351 description: str = "",
352 *,
353 return_type: type[_ReturnType],
354 args: _ReturnType | None = None,
355 kwargs: dict[str, Any] | None = None,
356 ) -> _ReturnType: ...
358 @overload
359 def emit[_ReturnType](
360 self,
361 name: str,
362 description: str = "",
363 *,
364 return_type: None = None,
365 args: _ReturnType,
366 kwargs: dict[str, Any] | None = None,
367 ) -> _ReturnType: ...
369 @overload
370 def emit(
371 self,
372 name: str,
373 description: str = "",
374 *,
375 return_type: None = None,
376 args: None = None,
377 kwargs: dict[str, Any] | None = None,
378 ) -> None: ...
380 def emit[_ReturnType](
381 self,
382 name: str,
383 description: str = "",
384 *,
385 return_type: type[_ReturnType] | None = None,
386 args: _ReturnType | None = None,
387 kwargs: dict[str, Any] | None = None,
388 ) -> _ReturnType | None:
389 """
390 Emit a signal, calling handlers in priority order.
392 Each handler transforms the first argument and passes it to the next handler.
394 Args:
395 name: Signal name
396 description: Optional description for new signals
397 return_type: Optional return type for new signals
398 args: List of positional arguments (first one is transformed through the chain)
399 kwargs: Keyword arguments passed to all handlers
401 Returns:
402 The transformed first argument after all handlers have processed it
404 """
405 if not (signal := self.get(name)):
406 signal = self.create(name, description, return_type)
408 arg_tuple = (args,)
409 kwargs = kwargs or {}
410 return signal.emit(*arg_tuple, **kwargs)
412 def connect(self, name: str, handler: Callable[..., Any], priority: int = SignalPriority.NORMAL) -> None:
413 """
414 Connect a handler to a signal, or queue it if the signal is not yet registered.
416 Args:
417 name: The signal name.
418 handler: The handler function to connect.
419 priority: The priority level for this handler (lower numbers execute first
421 """
422 if signal := self.get(name):
423 signal.connect(handler, priority)
424 else:
425 self.queue_action("connect", name, handler, priority)
427 def disconnect(self, name: str, handler: Callable[..., Any]) -> None:
428 """
429 Disconnect a handler from a signal, or queue it if the signal is not yet registered.
431 Args:
432 name: The signal name.
433 handler: The handler function to disconnect.
435 """
436 if signal := self.get(name):
437 signal.disconnect(handler)
438 else:
439 self.queue_action("disconnect", name, handler)
441 def disable(self, name: str, handler: Callable[..., Any]) -> None:
442 """
443 Temporarily disable a handler for a signal, or queue it if the signal is not yet registered.
445 Args:
446 name: The signal name.
447 handler: The handler function to disable
449 """
450 if signal := self.get(name):
451 signal.disable(handler)
452 else:
453 self.queue_action("disable", name, handler)
455 def enable(self, name: str, handler: Callable[..., Any]) -> None:
456 """
457 Enable a previously disabled handler, or queue it if the signal is not yet registered.
459 Args:
460 name: The signal name.
461 handler: The handler function to enable.
463 """
464 if signal := self.get(name):
465 signal.enable(handler)
466 else:
467 self.queue_action("enable", name, handler)
469 def is_queued(self, action: ActionType, name: str, handler: Callable[..., Any]) -> bool:
470 """
471 Check if a handler is queued for a signal action.
473 Args:
474 action: The action to check (connect, disconnect, disable, enable).
475 name: The signal name.
476 handler: The handler function to check.
478 Returns:
479 True if the handler is queued, False otherwise.
481 """
482 for queued_handler in self._queue[action].get(name, set()):
483 # Handle "connect" case where queued_handler is a tuple (handler, priority)
484 if isinstance(queued_handler, tuple):
485 if queued_handler[0] == handler:
486 return True
487 elif queued_handler == handler:
488 return True
489 return False
492registry = SignalRegistry.get_instance()