Event Module

Event module.

class pepperpy.event.Event(name, data=None, metadata=<factory>)[source]

Bases: object

Event class.

Parameters:

name (str)
data (Any)
metadata (Dict[str, Any])

name: str

data: Any = None

metadata: Dict[str, Any]

class pepperpy.event.EventBus(config=None)[source]

Bases: BaseModule[EventBusConfig]

Event bus implementation.

Parameters:: config (EventBusConfig | None)

__init__(config=None)[source]

Initialize event bus.

Parameters:: config (Optional[EventBusConfig]) – Optional event bus configuration
Return type:: None

async add_listener(event_name, handler, priority=0)[source]

Add event listener.

Parameters:

event_name (str) – Event name
handler (Callable[[Event], Awaitable[None]]) – Event handler
priority (int) – Handler priority (higher priority handlers are called first)

Raises:

EventError – If handler is not async or max listeners reached

Return type:

None

async remove_listener(event_name, handler)[source]

Remove event listener.

Parameters:

event_name (str) – Event name
handler (Callable[[Event], Awaitable[None]]) – Event handler

Return type:

None

async add_middleware(middleware)[source]

Add event middleware.

Parameters:: middleware (EventMiddleware) – Event middleware
Return type:: None

async remove_middleware(middleware)[source]

Remove event middleware.

Parameters:: middleware (EventMiddleware) – Event middleware
Return type:: None

async emit(event)[source]

Emit event.

Parameters:: event (Event) – Event to emit
Return type:: None

get_listeners(event_name)[source]

Get event listeners.

Parameters:: event_name (str) – Event name
Return type:: List[EventListener]
Returns:: List of event listeners

get_stats()[source]

Get event bus stats.

Return type:: Dict[str, Any]
Returns:: Event bus stats

async clear()[source]

Clear all event handlers and middleware.

Return type:: None

class pepperpy.event.EventBusConfig(name='event_bus', metadata=<factory>, max_listeners=None)[source]

Bases: ModuleConfig

Event bus configuration.

Parameters:

name (str)
metadata (Dict[str, Any])
max_listeners (int | None)

name: str = 'event_bus'

max_listeners: Optional[int] = None

metadata: Dict[str, Any]

exception pepperpy.event.EventError(message, context=None, cause=None, event_type=None, event_id=None)[source]

Bases: PepperpyError

Event error.

Parameters:

message (str)
context (Dict[str, Any] | None)
cause (Exception | None)
event_type (str | None)
event_id (str | None)

Return type:

None

__init__(message, context=None, cause=None, event_type=None, event_id=None)[source]

Initialize event error.

Parameters:

message (str) – Error message
context (Optional[Dict[str, Any]]) – Error context
cause (Optional[Exception]) – Error cause
event_type (Optional[str]) – Event type
event_id (Optional[str]) – Event ID

Return type:

None

class pepperpy.event.EventListener(event_name, handler, priority=0)[source]

Bases: object

Event listener class.

Parameters:

event_name (str)
handler (Callable[[Event], Awaitable[None]])
priority (int)

event_name: str

handler: Callable[[Event], Awaitable[None]]

priority: int = 0

class pepperpy.event.EventMiddleware(*args, **kwargs)[source]

Bases: Protocol

Event middleware protocol.

async before_event(event)[source]

Called before event is processed.

Parameters:: event (Event) – Event being processed
Return type:: None

async after_event(event)[source]

Called after event is processed.

Parameters:: event (Event) – Event that was processed
Return type:: None