Events API¶

events ¶

Core components for the library's event-driven architecture.

Event `dataclass` ¶

Encapsulate system event data.

to_dict ¶

to_dict() -> dict[str, Any]

Serialize the event to a dictionary.

EventEmitter ¶

Manage event subscriptions and dispatching.

clear_history ¶

clear_history() -> None

Purge the event history buffer.

close `async` ¶

close() -> None

Terminate processing and clear listeners.

emit `async` ¶

emit(*, event_type: EventType | str, data: EventData | None = None, source: Any = None) -> None

Dispatch an event to registered listeners.

emit_nowait ¶

emit_nowait(*, event_type: EventType | str, data: EventData | None = None, source: Any = None) -> None

Schedule an event dispatch without awaiting.

get_event_history ¶

get_event_history(*, event_type: EventType | str | None = None, limit: int = 100) -> list[Event]

Retrieve the recorded event history.

get_stats ¶

get_stats() -> dict[str, Any]

Retrieve internal statistics.

listener_count ¶

listener_count(*, event_type: EventType | str) -> int

Return the number of listeners for an event.

listeners ¶

listeners(*, event_type: EventType | str) -> list[EventHandler]

Return the list of handlers for an event.

off ¶

off(*, event_type: EventType | str, handler: EventHandler | None = None) -> None

Unregister specific or all handlers for an event.

off_any ¶

off_any(*, handler: EventHandler | None = None) -> None

Unregister wildcard handlers.

on ¶

on(*, event_type: EventType | str, handler: EventHandler) -> None

Register a persistent event handler.

on_any ¶

on_any(*, handler: EventHandler) -> None

Register a wildcard handler for all events.

once ¶

once(*, event_type: EventType | str, handler: EventHandler) -> None

Register a one-time event handler.

pause ¶

pause() -> None

Suspend event processing.

remove_all_listeners ¶

remove_all_listeners(*, event_type: EventType | str | None = None) -> None

Clear listeners for a specific or all events.

resume ¶

resume() -> asyncio.Task[None] | None

Resume event processing and flush queue.

set_max_listeners ¶

set_max_listeners(*, max_listeners: int) -> None

Update the maximum listener limit.

wait_for `async` ¶

wait_for(*, event_type: EventType | str | list[EventType | str], condition: Callable[[Event], bool] | None = None, timeout: float | None = None) -> Event

Await the emission of a specific event.