Skip to content

Events API

ALL_COMPLETED module-attribute 

ALL_COMPLETED: Final[Literal['ALL_COMPLETED']] = asyncio.ALL_COMPLETED

FIRST_COMPLETED module-attribute 

FIRST_COMPLETED: Final[Literal['FIRST_COMPLETED']] = asyncio.FIRST_COMPLETED

FIRST_EXCEPTION module-attribute 

FIRST_EXCEPTION: Final[Literal['FIRST_EXCEPTION']] = asyncio.FIRST_EXCEPTION

__all__ module-attribute 

__all__ = [
    "ALL_COMPLETED",
    "FIRST_COMPLETED",
    "FIRST_EXCEPTION",
    "Event",
    "EventBus",
    "EventResult",
    "EventState",
    "EventStatesGroup",
    "ScheduleEvent",
    "pub",
    "sub",
]

EventBus

__brokers instance-attribute 

__brokers = (InMemoryBroker(),)

__middleware instance-attribute 

__middleware = [(item, item.bind(self)) for item in middleware or ()]

__schedules instance-attribute 

__schedules: list[tuple[str | timedelta, Callable[[], Awaitable[Any]]]] = []

__init__ 

__init__(
    *brokers: Brokerable, middleware: Sequence[EventsMiddleware] | None = None
)

get async 

get[E: Eventable](event_type: type[E]) -> AsyncIterator[EventState[E]]

get[E: Eventable](event: E) -> EventState[E] | None

get(*events: *ETs) -> EventStatesGroup[*ETs,]

get(*items: Any) -> Any

subscribe 

subscribe[E: Event](
    event_type: type[E],
    handler: BaseNullableAction[E, Any]
    | BaseAction[E, Any]
    | BaseSequenceAction[E, Any],
) -> None

retract async 

retract[E: Eventable](state: EventState[E]) -> bool

publish async 

publish[E: Eventable](event: E) -> EventState[E]

publish(*events: *ETs) -> EventStatesGroup[*ETs,]

publish(*events: *ETs) -> EventStatesGroup | EventState

__run_middleware async 

__run_middleware(state: EventState[Eventable]) -> None

as_completed async 

as_completed[E: Eventable](
    event_states_group: EventStatesGroup[*tuple[E, ...],],
) -> AsyncIterator[EventState[E]]

__call__ 

__call__(*events: *ETs) -> EventGroupInBus[*ETs,]

schedule 

schedule(on: str | timedelta, action: Callable[[], Awaitable[Any]]) -> None

__run_schedule async staticmethod 

__run_schedule(
    on: str | timedelta, action: Callable[[], Awaitable[Any]]
) -> None

scheduler async 

scheduler()

add_middleware 

add_middleware[**P](
    middleware: type[EventsMiddlewareFactory[P]],
    *args: P.args,
    **kwargs: P.kwargs,
) -> None

Event

Bases: BaseSchema

model_config class-attribute instance-attribute 

model_config = ConfigDict(arbitrary_types_allowed=True)

__event_id__ property 

__event_id__: UUID

EventResult

state instance-attribute 

state = event_state

__init__ 

__init__(event_state: EventState[E])

__repr__ 

__repr__() -> str

add_pending 

add_pending(future: Future[Any]) -> None

__iter__ 

__iter__()

EventState

__event instance-attribute 

__event = event

__result instance-attribute 

__result = EventResult(self)

__seen_completions instance-attribute 

__seen_completions = 0

__seen_errors instance-attribute 

__seen_errors = 0

event property 

event: E

result property 

result: EventResult[E]

__init__ 

__init__(event: E) -> None

__repr__ 

__repr__() -> str

wait async 

wait(
    *,
    return_when: ReturnWhenLiteral = ALL_COMPLETED,
    timeout: int | None = None,
) -> EventResult[E]

EventStatesGroup

__event_states instance-attribute 

__event_states = event_states

__init__ 

__init__(*event_states: EventState[Eventable]) -> None

__iter__ 

__iter__()

__repr__ 

__repr__() -> str

ScheduleEvent

Bases: Event

__when__ class-attribute 

__when__: timedelta

pub 

pub(*, bus: EventBus = Depends(), on: str | timedelta) -> SchedulePubDecorator

pub(*, bus: EventBus = Depends(), on: None = None) -> PubDecorator

pub(
    *, bus: EventBus = Depends(), on: str | timedelta | None = None
) -> PubDecorator | SchedulePubDecorator

sub 

sub[T: Event, R: BaseSchema](
    *event_types: type[T], bus: EventBus = Depends()
) -> SubDecorator