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: tuple[Brokerable, ...]

middleware instance-attribute

middleware: list[Callable[[EventState[Eventable]], Awaitable[None]]] = [
    item.resolve(self) for item in middleware or ()
]

schedules instance-attribute

schedules: list[Callable[[], Awaitable[None]]] = []

__init__

__init__(
    *brokers: Brokerable,
    middleware: Sequence[MiddlewareBindable] | 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: NullableActionable[E, Any]
    | Actionable[E, Any]
    | SequenceActionable[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[Result](
    on: str | timedelta, action: Callable[[], Awaitable[Result]]
) -> None

run_schedule async staticmethod

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

scheduler async

scheduler() -> AsyncIterator[None]

add_middleware

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

Event

Bases: BaseSchema

model_config class-attribute

model_config: ConfigDict = ConfigDict(arbitrary_types_allowed=True)

__event_id__ property

__event_id__: UUID

from_transport classmethod

from_transport(payload: str, event_id: UUID) -> Self

Restore the sender's identity when decoding a transported event.

EventResult

Bases: tuple[list[R], list[Future[Any]]], Generic[E, R]

state instance-attribute

state: EventState[E] = event_state

schema instance-attribute

schema: type[R] | None = schema

success_events instance-attribute

success_events: list[R] = self[0]

pending_events instance-attribute

pending_events: list[Future[Any]] = self[1]

error_events instance-attribute

error_events: list[BaseException] = []

completion_count instance-attribute

completion_count: int = 0

seen_completions instance-attribute

seen_completions: int = 0

seen_errors instance-attribute

seen_errors: int = 0

views instance-attribute

views: dict[type[Schema], EventResult[E, Any]] = {}

errors property

errors: tuple[BaseException, ...]

__new__

__new__(event_state: EventState[E], schema: type[R] | None = None) -> Self

__init__

__init__(event_state: EventState[E], schema: type[R] | None = None) -> None

__eq__

__eq__(other: object) -> bool

__ne__

__ne__(other: object) -> bool

__hash__

__hash__() -> int

__repr__

__repr__() -> str

add_pending

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

route

route[S: Schema](schema: type[S]) -> EventResult[E, S]

select

select(value: Any) -> list[R]

complete

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

wait async

wait(
    *,
    return_when: ReturnWhenLiteral = ALL_COMPLETED,
    timeout: float | None = None,
) -> Self

completed

completed(
    return_when: ReturnWhenLiteral, seen_completions: int, seen_errors: int
) -> bool

__iter__

__iter__() -> Iterator[list[Any]]

EventState

event_value instance-attribute

event_value = event

event_result instance-attribute

event_result: EventResult[E] = EventResult(self)

event property

event: E

result property

result: EventResult[E]

__init__

__init__(event: E) -> None

__repr__

__repr__() -> str

wait async

wait[R: Schema](
    schema: type[R],
    *,
    return_when: ReturnWhenLiteral = ALL_COMPLETED,
    timeout: float | None = None,
) -> EventResult[E, R]

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

wait(
    schema: type[Schema] | None = None,
    *,
    return_when: ReturnWhenLiteral = ALL_COMPLETED,
    timeout: float | None = None,
) -> EventResult[E]

EventStatesGroup

event_states instance-attribute

event_states = event_states

__init__

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

__iter__

__iter__() -> Iterator[EventState[Eventable]]

__repr__

__repr__() -> str

ScheduleEvent

Bases: Event

__when__ class-attribute

__when__: timedelta

pub

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

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

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

sub

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

__all__ module-attribute

__all__ = ['RabbitMQBroker']

RabbitMQBroker

Bases: TransportBroker

native instance-attribute

native = native

route instance-attribute

route = route

subscription instance-attribute

subscription = subscription

__init__

__init__(
    native: RabbitBroker,
    *,
    route: Callable[[type[Eventable]], str] = event_route,
    subscription: Callable[
        [type[Event], Callable[[EventMessage], Awaitable[None]]],
        Callable[[], Awaitable[None]],
    ]
    | None = None,
) -> None

connect async

connect() -> None

register

register(
    event_type: type[Event], handler: Callable[[EventMessage], Awaitable[None]]
) -> Callable[[], Awaitable[None]]

send async

send(message: EventMessage, event_type: type[Eventable]) -> None

stop async

stop() -> None

__all__ module-attribute

__all__ = ['InMemoryBroker']

InMemoryBroker

Bases: Brokerable

Route events through an in-memory bubus bus.

bus instance-attribute

bus = bus

registered instance-attribute

registered: bool = False

handlers instance-attribute

handlers: list[tuple[type[Event], Callable[[Event], Awaitable[object]]]] = []

states instance-attribute

states: dict[UUID, EventState[Any]] = {}

tasks instance-attribute

tasks: dict[UUID, list[Task[Any]]] = {}

processing instance-attribute

processing: dict[UUID, Task[None]] = {}

starts instance-attribute

starts: dict[UUID, Future[None]] = {}

started instance-attribute

started: set[UUID] = set()

__init__

__init__(bus: EventBus | None = None) -> None

native_bus

native_bus() -> BubusEventBus

subscribe

subscribe[E: Event](
    event_type: type[E],
    handler: NullableActionable[E, Any]
    | Actionable[E, Any]
    | SequenceActionable[E, Any],
) -> None

dispatch async

dispatch[E: Eventable](event_state: EventState[E]) -> None

prepare

prepare[E: Eventable](event_state: EventState[E]) -> None

enqueue

enqueue[E: Eventable](state: EventState[E]) -> None

process async

process(envelope: Any) -> None

deliver async

deliver(event_id: UUID, *, raise_errors: bool = False) -> 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

retract async

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

stop async

stop() -> None

__all__ module-attribute

__all__ = ['AsyncQueueBroker']

AsyncQueueBroker

Bases: InMemoryBroker

Schedule concurrent event delivery through a local asyncio queue.

queue instance-attribute

queue: Queue[UUID] = asyncio.Queue[UUID]() if queue is None else queue

consumer instance-attribute

consumer: Task[None] | None = None

__init__

__init__(queue: Queue[UUID] | None = None) -> None

dispatch async

dispatch[E: Eventable](event_state: EventState[E]) -> None

consume async

consume() -> None

complete

complete(event_id: UUID, _: Task[None]) -> None

stop async

stop() -> None

__all__ module-attribute

__all__ = ['NATSBroker']

NATSBroker

Bases: TransportBroker

native instance-attribute

native = native

route instance-attribute

route = route

subscription instance-attribute

subscription = subscription

__init__

__init__(
    native: NatsBroker,
    *,
    route: Callable[[type[Eventable]], str] = event_route,
    subscription: Callable[
        [type[Event], Callable[[EventMessage], Awaitable[None]]],
        Callable[[], Awaitable[None]],
    ]
    | None = None,
) -> None

connect async

connect() -> None

register

register(
    event_type: type[Event], handler: Callable[[EventMessage], Awaitable[None]]
) -> Callable[[], Awaitable[None]]

send async

send(message: EventMessage, event_type: type[Eventable]) -> None

stop async

stop() -> None

__all__ module-attribute

__all__ = ['KafkaBroker']

KafkaBroker

Bases: TransportBroker

native instance-attribute

native = native

route instance-attribute

route = route

group_id instance-attribute

group_id = group_id

subscription instance-attribute

subscription = subscription

__init__

__init__(
    native: KafkaBroker,
    *,
    route: Callable[[type[Eventable]], str] = event_route,
    group_id: str | None = "fastsprout",
    subscription: Callable[
        [type[Event], Callable[[EventMessage], Awaitable[None]]],
        Callable[[], Awaitable[None]],
    ]
    | None = None,
) -> None

connect async

connect() -> None

register

register(
    event_type: type[Event], handler: Callable[[EventMessage], Awaitable[None]]
) -> Callable[[], Awaitable[None]]

send async

send(message: EventMessage, event_type: type[Eventable]) -> None

stop async

stop() -> None