events

Provides models and utilities for handling events.

Event

Bases: PlugboardBaseModel, ABC

Event is a base model for all events.

Attributes:

Name	Type	Description
`type`	`str`	Type of the `Event`.
`id`	`UUID4`	UUID v4 unique identifier for the `Event`.
`timestamp`	`UTCDateTime`	UTC timestamp for the `Event`.
`source`	`str`	Source of the `Event`.
`version`	`str`	Version of the `Event`.
`data`	`dict[str, Any] \| BaseModel`	Data associated with the `Event`.
`metadata`	`dict[str, str]`	Metadata for the `Event`.

handler `classmethod`

handler(method: AsyncCallable) -> AsyncCallable

Registers a class method as an event handler.

safe_type `classmethod`

safe_type(event_type: Optional[str] = None) -> str

Returns a safe event type string for use in broker topic strings.

SystemEvent

Bases: Event, ABC

SystemEvent is a base model for system events.

Attributes:

Name	Type	Description
`type`	`str`	Type of the `SystemEvent`.
`id`	`UUID4`	UUID v4 unique identifier for the `SystemEvent`.
`timestamp`	`UTCDateTime`	UTC timestamp for the `SystemEvent`.
`source`	`str`	Source of the `SystemEvent`.
`version`	`str`	Version of the `SystemEvent`.
`data`	`dict[str, Any] \| BaseModel`	Data associated with the `SystemEvent`.
`metadata`	`dict[str, str]`	Metadata for the `SystemEvent`.

StopEvent

Bases: SystemEvent

StopEvent is a system event to stop the application.

EventConnectorBuilder

EventConnectorBuilder(connector_builder: ConnectorBuilder)

EventConnectorBuilder constructs connectors for component event handlers.

build

build(
    components: Iterable[Component],
) -> dict[str, Connector]

Returns mapping of connectors for events handled by components.

EventHandlers

EventHandlers provides a decorator for registering event handlers.

add `classmethod`

add(
    event: Type[Event] | Event,
) -> _t.Callable[[AsyncCallable], AsyncCallable]

Decorator that registers class methods as handlers for specific event types.

Parameters:

Name	Type	Description	Default
`event`	`Type[Event] \| Event`	Event class this handler processes	required

Returns:

Name	Type	Description
`Callable`	`Callable[[AsyncCallable], AsyncCallable]`	Decorated method

get `classmethod`

get(
    _class: Type, event: Type[Event] | Event
) -> AsyncCallable

Retrieve a handler for a specific class and event type.

Parameters:

Name	Type	Description	Default
`_class`	`Type`	Class to handle event for	required
`event`	`Type[Event] \| Event`	Event class or instance to handle	required

Returns:

Name	Type	Description
`Callable`	`AsyncCallable`	The event handler method

Raises:

Type	Description
`KeyError`	If no handler found for class or event type