pywry.callbacks

Callback registry for handling widget events.

---

Callback Types

pywry.callbacks.CallbackFunc module-attribute

CallbackFunc = Callable[..., None] | Callable[..., Awaitable[None]]

pywry.callbacks.WidgetType

Bases: str, Enum

Widget types for event routing.

---

Callback Registry

pywry.callbacks.get_registry

get_registry() -> CallbackRegistry

Get the global callback registry instance.

Returns:

Type	Description
CallbackRegistry	The callback registry singleton.

pywry.callbacks.CallbackRegistry

CallbackRegistry()

Registry for managing event callbacks.

Supports both sync and async callbacks with namespace:event-name pattern. Thread-safe singleton pattern.

Initialize the registry.

Attributes

_instance class-attribute instance-attribute

_instance: CallbackRegistry | None = None

_initialized class-attribute instance-attribute

_initialized: bool = True

_callbacks instance-attribute

_callbacks: dict[str, dict[str, list[CallbackFunc]]] = {}

_scoped_callbacks instance-attribute

_scoped_callbacks: dict[str, dict[str, list[tuple[str, str, CallbackFunc]]]] = {}

_destroyed_labels instance-attribute

_destroyed_labels: set[str] = set()

_executor instance-attribute

_executor: ThreadPoolExecutor | None = None

_max_workers instance-attribute

_max_workers: int = 4

_pending_futures instance-attribute

_pending_futures: list[Future[None]] = []

Functions

__new__

__new__() -> CallbackRegistry

Create or return the singleton instance.

_matches staticmethod

_matches(pattern: str, source: str) -> bool

Check if a pattern matches a source string.

Supports wildcards (*) for flexible matching.

Parameters:

Name	Type	Description	Default
pattern	str	Pattern to match (can contain * wildcards).	required
source	str	Source string to match against.	required

Returns:

Type	Description
bool	True if pattern matches source.

Examples:

>>> CallbackRegistry._matches("*", "anything")
True
>>> CallbackRegistry._matches("grid_*", "grid_123")
True
>>> CallbackRegistry._matches("chart", "chart")
True

register

register(label: str, event_type: str, handler: CallbackFunc, widget_type: str = '*', widget_id: str = '*') -> bool

Register an event handler.

Parameters:

Name	Type	Description	Default
label	str	The window label.	required
event_type	str	The event type (namespace:event-name or * for wildcard).	required
handler	CallbackFunc	The callback function.	required
widget_type	str	Widget type filter ("grid", "chart", "toolbar", "html", "window", or "*").	'*'
widget_id	str	Widget ID filter (specific ID or "*" for all).	'*'

Returns:

Type	Description
bool	True if registered successfully, False otherwise.

unregister

unregister(label: str, event_type: str | None = None, handler: CallbackFunc | None = None) -> bool

Unregister event handler(s).

Parameters:

Name	Type	Description	Default
label	str	The window label.	required
event_type	str or None	The event type (None to unregister all for this label).	None
handler	CallbackFunc or None	Specific handler to remove (None to remove all for event_type).	None

Returns:

Type	Description
bool	True if any handlers were removed, False otherwise.

_collect_simple_handlers

_collect_simple_handlers(label: str, event_type: str) -> list[CallbackFunc]

Collect handlers from simple (non-scoped) callback structure.

_collect_scoped_handlers

_collect_scoped_handlers(label: str, event_type: str, widget_type: str, widget_id: str) -> list[CallbackFunc]

Collect handlers from scoped callback structure with pattern matching.

_drain

_drain(timeout: float | None = None) -> None

Wait for all pending sync handler futures to complete.

_ensure_executor

_ensure_executor() -> ThreadPoolExecutor

Lazily create the thread pool executor for sync handlers.

_invoke_handler

_invoke_handler(handler: CallbackFunc, data: Any, event_type: str, label: str) -> bool

Invoke a single handler with appropriate arguments.

Sync handlers are submitted to a thread pool so they never block the reader thread. Async handlers are scheduled via the BlockingPortal for proper async runtime integration.

_invoke_async_handler

_invoke_async_handler(handler: CallbackFunc, data: Any, event_type: str, label: str, num_params: int) -> bool

Invoke an async handler via the BlockingPortal.

Uses portal.start_task_soon() to schedule the coroutine without blocking the current thread. Errors are logged via log_callback_error.

dispatch

dispatch(label: str, event_type: str, data: Any) -> bool

Dispatch an event to registered handlers.

Matches handlers based on event type and optional widget_type/widget_id from the event data. Scoped handlers are matched using wildcard patterns.

Parameters:

Name	Type	Description	Default
label	str	The window label.	required
event_type	str	The event type.	required
data	Any	The event data. May contain widget_type and widget_id/gridId/chartId.	required

Returns:

Type	Description
bool	True if any handlers were called, False otherwise.

destroy

destroy(label: str) -> bool

Completely destroy all resources for a window label.

This removes all callbacks and marks the label as destroyed to prevent future registrations.

Parameters:

Name	Type	Description	Default
label	str	The window label to destroy.	required

Returns:

Type	Description
bool	True if the label was destroyed, False if it didn't exist.

recover_label

recover_label(label: str) -> bool

Recover a destroyed label to allow new registrations.

This is useful when restarting the application or reusing a label that was previously destroyed.

Parameters:

Name	Type	Description	Default
label	str	The window label to recover.	required

Returns:

Type	Description
bool	True if the label was recovered (was in destroyed set).

is_destroyed

is_destroyed(label: str) -> bool

Check if a window label has been destroyed.

Parameters:

Name	Type	Description	Default
label	str	The window label.	required

Returns:

Type	Description
bool	True if destroyed, False otherwise.

has_handlers

has_handlers(label: str) -> bool

Check if a window has any registered handlers.

Parameters:

Name	Type	Description	Default
label	str	The window label.	required

Returns:

Type	Description
bool	True if handlers exist, False otherwise.

get_labels

get_labels() -> list[str]

Get all window labels with registered handlers.

Returns:

Type	Description
List of window labels.

clear

clear() -> None

Clear all callbacks and destroyed labels.

Use with caution - primarily for testing.