Signals

This section contains documentation for the signals module.

METADATA:

File: signals.py: Project: paperap
Created: 2025-03-09: Version: 0.0.9

Author: Jess Mann Email: jess@jmann.me

Copyright (c) 2025 Jess Mann

LAST MODIFIED:

2025-03-09 By Jess Mann

class paperap.signals.QueueType[source]

Bases: TypedDict

A type used by SignalRegistry for storing queued signal actions.

connect: dict[str, set[tuple[Callable[..., Any], int]]]

disconnect: dict[str, set[Callable[..., Any]]]

disable: dict[str, set[Callable[..., Any]]]

enable: dict[str, set[Callable[..., Any]]]

final class paperap.signals.SignalPriority[source]

Bases: object

Priority levels for signal handlers.

Any int can be provided, but these are the recommended values.

FIRST = 0

HIGH = 25

NORMAL = 50

LOW = 75

LAST = 100

class paperap.signals.SignalParams[source]

Bases: TypedDict

A type used by SignalRegistry for storing signal parameters.

name: str

description: str

class paperap.signals.Signal(name, description='')[source]

Bases: Generic

A signal that can be connected to and emitted.

Handlers can be registered with a priority to control execution order. Each handler receives the output of the previous handler as its first argument, enabling a filter/transformation chain.

Parameters:

name (str)
description (str)

__init__(name, description='')[source]

Parameters:

name (str)
description (str)

name: str

description: str

connect(handler, priority=50)[source]

Connect a handler to this signal.

Parameters:

handler (Callable[…, _ReturnType]) – The handler function to be called when the signal is emitted.
priority (int) – The priority level for this handler (lower numbers execute first).

Return type:

None

disconnect(handler)[source]

Disconnect a handler from this signal.

Parameters:: handler (Callable[…, _ReturnType]) – The handler to disconnect.
Return type:: None

emit(*args, **kwargs)[source]

Emit the signal, calling all connected handlers in priority order.

Each handler receives the output of the previous handler as its first argument. Other arguments are passed unchanged.

Parameters:

*args (Any) – Positional arguments to pass to handlers.
**kwargs (Any) – Keyword arguments to pass to handlers.

Return type:

_ReturnType | None

Returns:

The final result after all handlers have processed the data.

disable(handler)[source]

Temporarily disable a handler without disconnecting it.

Parameters:: handler (Callable[…, _ReturnType]) – The handler to disable.
Return type:: None

enable(handler)[source]

Re-enable a temporarily disabled handler.

Parameters:: handler (Callable[…, _ReturnType]) – The handler to enable.
Return type:: None

class paperap.signals.SignalRegistry[source]

Bases: object

Registry of all signals in the application.

Signals can be created, connected to, and emitted through the registry.

Examples

>>> SignalRegistry.emit(
...     "document.save:success",
...     "Fired when a document has been saved successfully",
...     kwargs = {"document": document}
... )

>>> filtered_data = SignalRegistry.emit(
...     "document.save:before",
...     "Fired before a document is saved. Optionally filters the data that will be saved.",
...     args = (data,),
...     kwargs = {"document": document}
... )

>>> SignalRegistry.connect("document.save:success", my_handler)

__init__()[source]

static __new__(cls)[source]

Ensure that only one instance of the class is created.

Return type:: Self
Returns:: The singleton instance of this class.

classmethod get_instance()[source]

Get the singleton instance of this class.

Return type:: Self
Returns:: The singleton instance of this class.

register(signal)[source]

Register a signal and process queued actions.

Parameters:: signal (Signal[Any]) – The signal to register.
Return type:: None

queue_action(action, name, handler, priority=None)[source]

Queue any signal-related action to be processed when the signal is registered.

Parameters:

action (Literal['connect', 'disconnect', 'disable', 'enable']) – The action to queue (connect, disconnect, disable, enable).
name (str) – The signal name.
handler (Callable[..., Any]) – The handler function to queue.
priority (int | None) – The priority level for this handler (only for connect action).

Raises:

ValueError – If the action is invalid.

Return type:

None

get(name)[source]

Get a signal by name.

Parameters:: name (str) – The signal name.
Return type:: Optional[Signal[Any]]
Returns:: The signal instance, or None if not found.

list_signals()[source]

List all registered signal names.

Return type:: list[str]
Returns:: A list of signal names.

create(name, description='', return_type=None)[source]

Create and register a new signal.

Parameters:

name (str) – Signal name
description (str) – Optional description for new signals
return_type (type[TypeVar(R)] | None) – Optional return type for new signals

Return type:

Signal[TypeVar(R)]

Returns:

The new signal instance.

emit(name, description='', *, return_type=None, args=None, kwargs=None)[source]

Emit a signal, calling handlers in priority order.

Each handler transforms the first argument and passes it to the next handler.

Parameters:

name (str) – Signal name
description (str) – Optional description for new signals
return_type (type[TypeVar(_ReturnType)] | None) – Optional return type for new signals
args (Optional[TypeVar(_ReturnType)]) – List of positional arguments (first one is transformed through the chain)
kwargs (dict[str, Any] | None) – Keyword arguments passed to all handlers

Return type:

Optional[TypeVar(_ReturnType)]

Returns:

The transformed first argument after all handlers have processed it

connect(name, handler, priority=50)[source]

Connect a handler to a signal, or queue it if the signal is not yet registered.

Parameters:

name (str) – The signal name.
handler (Callable[..., Any]) – The handler function to connect.
priority (int) – The priority level for this handler (lower numbers execute first

Return type:

None

disconnect(name, handler)[source]

Disconnect a handler from a signal, or queue it if the signal is not yet registered.

Parameters:

name (str) – The signal name.
handler (Callable[..., Any]) – The handler function to disconnect.

Return type:

None

disable(name, handler)[source]

Temporarily disable a handler for a signal, or queue it if the signal is not yet registered.

Parameters:

name (str) – The signal name.
handler (Callable[..., Any]) – The handler function to disable

Return type:

None

enable(name, handler)[source]

Enable a previously disabled handler, or queue it if the signal is not yet registered.

Parameters:

name (str) – The signal name.
handler (Callable[..., Any]) – The handler function to enable.

Return type:

None

is_queued(action, name, handler)[source]

Check if a handler is queued for a signal action.

Parameters:

action (Literal['connect', 'disconnect', 'disable', 'enable']) – The action to check (connect, disconnect, disable, enable).
name (str) – The signal name.
handler (Callable[..., Any]) – The handler function to check.

Return type:

bool

Returns:

True if the handler is queued, False otherwise.