xtr-event-dispatcher

Let the parts of an application talk through events: listeners and subscribers run by priority, any of them able to stop the rest.

Why?

When placing an order has to send a receipt, update stock and write an audit line, the code that places it should not call all three. It dispatches an OrderPlaced event; each concern listens for it on its own, in the order their priorities give, and any listener can stop the ones after it. Adding a fourth concern is adding a listener, not editing the order code.

• 📣 Async dispatch — listeners are plain functions or async def, awaited one after another.
• 🔢 Priorities and stopping — higher runs first; an event a listener stops goes no further.
• 🧾 Subscribers — a class declaring every event it listens to, in one place.
• 💤 Lazy listeners — the object a listener belongs to is built when an event first reaches it.
• 🔒 Immutable, scoped, compiled — hand a dispatcher out without letting anyone change it, and add listeners for one scope without touching a shared one.
• 🧩 A bundle — @as_event_listener on a class, a method or a function, and the container wires it, ordered by priority and before/after.
• 🔍 Tracing — which listeners ran, which did not, and which events nobody heard.

Install

uv add xtr-event-dispatcher

With a container:

uv add "xtr-event-dispatcher[di]"
uv add "xtr-event-dispatcher[di,console]"  # + debug:event-dispatcher, in debug mode

A library that only dispatches events depends on xtr-event-dispatcher-contracts instead, and leaves the dispatcher to the application.

Quick start

from dataclasses import dataclass

from xtr_event_dispatcher import Event, EventDispatcher


@dataclass(frozen=True)
class OrderPlaced(Event):
    order_id: int


async def send_receipt(event: OrderPlaced) -> None: ...


def reject_fraud(event: OrderPlaced) -> None:
    if is_fraudulent(event.order_id):
        event.stop_propagation()  # send_receipt will not run


dispatcher = EventDispatcher()
dispatcher.add_listener(OrderPlaced, reject_fraud, priority=100)
dispatcher.add_listener(OrderPlaced, send_receipt)

await dispatcher.dispatch(OrderPlaced(42))

Events and names

An event is any object; derive from Event to let listeners stop it. Events are keyed by name: a string, or a class standing for "<module>.<qualname>". dispatch(event) without a name uses the event's class, so add_listener(OrderPlaced, ...) and dispatch(OrderPlaced(42)) meet. A listener of a base class does not hear its subclasses: a package wanting a broad audience dispatches one event class per situation.

GenericEvent is an event without a class of its own: a subject, and arguments read like a mapping.

event = await dispatcher.dispatch(GenericEvent(order, {"notify": True}), "order.saved")
if event["notify"]:
    ...

Listeners

A listener is called with up to three arguments — the event, the name it was dispatched under, and the dispatcher — as many as it takes, by position. What it returns is ignored, except that an awaitable is awaited before the next listener runs. Adding one that requires more raises ListenerSignatureError there and then.

def log(event: OrderPlaced) -> None: ...
def route(event: Event, name: str) -> None: ...
async def chain(event: Event, name: str, dispatcher: EventDispatcherInterface) -> None: ...

Listeners run by priority, highest first; sharing a priority, in the order added. Bound methods of one object compare equal, so remove_listener(event, obj.method) removes what add_listener(event, obj.method) added. A listener added while its event is being dispatched first runs on the next dispatch; one removed meanwhile does not run for the rest of it.

Subscribers

class OrderMailer(EventSubscriberInterface):
    @classmethod
    @override
    def get_subscribed_events(cls) -> Mapping[str | type, SubscribedEvents]:
        return {
            OrderPlaced: "on_placed",  # priority 0
            OrderShipped: ("on_shipped", 10),  # (method, priority)
            OrderCancelled: [  # several listeners
                "notify_customer",
                {"method": "notify_warehouse", "priority": -5},
            ],
        }


dispatcher.add_subscriber(OrderMailer())

A mapping may also carry before/after (see below); a dispatcher on its own cannot honour them, so it requires a priority beside them and ignores them — a container orders by them.

Lazy listeners

dispatcher.add_listener(OrderPlaced, LazyListener(build_mailer, "on_placed"))

build_mailer is awaited just before the listener first runs — never, when an earlier listener stops the event — and what it built is kept. Until then get_listeners returns the LazyListener; afterwards, the bound method.

Dispatchers

A listener receives the dispatcher that ran it: the scoped, compiled or traceable one itself, but the wrapped one through an ImmutableEventDispatcher, which dispatches by delegating.

A TraceableEventDispatcher times every listener: each ListenerInfo of get_called_listeners() carries its calls and their total duration, in seconds, and every "Notified" debug record a duration. Wrapping an EventDispatcher — the container's always are — it leaves running the listeners to it and only watches each call.

ImmutableEventDispatcher and ScopedEventDispatcher wrap an IntrospectableDispatcherInterface: a dispatcher that dispatches and reads its listeners, nothing more — so one written against the contracts alone can be wrapped too.

A trace per unit of work

A TraceableEventDispatcher records into one trace per instance, growing until reset(). When several units of work overlap — concurrent requests through one shared dispatcher — that single trace would mix them. So it also offers begin_unit() and end_unit(): begin_unit() opens a trace scoped to the calling context, end_unit() closes it and recording returns to the instance. Because a unit lives in a context variable, overlapping units each record only their own, and — the trace being mutated in place — a synchronous listener run in a copied context still records into the unit the context points at.

A unit of work exists only when tracing is on: the trace is a development instrument, so the bundle wraps a dispatcher in a TraceableEventDispatcher only in debug mode, and only then do begin_unit/end_unit do anything. A caller framing each request as a unit — such as xtr-http-kernel's lifecycle — calls them when they are present and leaves them alone otherwise.

Use in an application

Everything adding this package to an application on xtr-dependency-injection takes — and, read backwards, what removing it undoes.

• Install — uv add "xtr-event-dispatcher[di]"; console adds debug:event-dispatcher, registered in debug mode only.
• Recipe — uv run xtr-recipes recipes:sync does the Activate step below: it lists EventDispatcherBundle, or leaves it out when the messenger or scheduler bundle already requires it. There is no config file, environment or ignore line to write.
• Activate — EventDispatcherBundle: {"all": True} in BUNDLES in <app>/bundles.py, imported from xtr_event_dispatcher.bundle. The messenger and scheduler bundles require it when it is installed.
• Brings along — the logging bundle, when xtr-logging is installed.
• Configure — optional: with no configuration there is one empty dispatcher, and every scanned listener joins it. Aliases and named dispatchers go in <app>/config/event_dispatcher.py, a @configure function returning EventDispatcherConfig — see Configuration.
• Environment — nothing.
• Ignore — nothing.
• Remove — drop the BUNDLES entry and every @as_event_listener, delete <app>/config/event_dispatcher.py, then uv remove xtr-event-dispatcher.
• Check — debug:bundles shows event_dispatcher as listed or required, and active; in debug mode, debug:event-dispatcher lists every listener the scan found.

Kernel / bundle

# app/bundles.py
from xtr_event_dispatcher.bundle import EventDispatcherBundle

BUNDLES = {EventDispatcherBundle: {"all": True}}

The application gets a dispatcher under EventDispatcherInterface — the contract's and this package's — and ListenerIntrospectionInterface, holding every listener its scan finds:

from xtr_dependency_injection import Injected
from xtr_event_dispatcher import as_event_listener


@as_event_listener()  # the event is the first parameter's type
async def send_receipt(event: OrderPlaced, mailer: Injected[Mailer]) -> None: ...


class Stock:  # built when an event first reaches it
    def __init__(self, repository: StockRepository) -> None: ...

    @as_event_listener(priority=10)
    def reserve(self, event: OrderPlaced | OrderEdited) -> None: ...


@as_event_listener(OrderShipped)  # calls on_order_shipped, else __call__
class Notifier:
    def on_order_shipped(self, event: OrderShipped) -> None: ...


class Audit(EventSubscriberInterface): ...  # every subscriber is registered

@as_event_listener(event=None, *, method=None, priority=None, dispatcher=None, before=None, after=None), repeatable, on a class, a method — static and class methods included, above or below their decorator — or a function:

• event — a name or a class. Without it, the first parameter's annotation — each member of a union — is the event; the base Event does not count. Annotations must be importable at runtime.
• method — on a class only. Without it, the class is called through __call__ when the event is read from its signature, else on_<event> (on_order_placed for OrderPlaced or "order.placed"), then __call__. On a method it is an error: the method is the one called.
• before / after — classes, functions or methods, or "module:Qualified.name" strings; a callable object is refused. A listener with a priority keeps it and is only reordered among its equals; one without takes the priority its place needs. A target naming something not installed is ignored; a Class.method whose class listens through another method is an error. An inherited method is also known by the class defining it.
• dispatcher — a named dispatcher from EventDispatcherConfig.dispatchers, injected with Annotated[EventDispatcherInterface, Target("audit")].

A function's own parameters come first; the ones the container fills — Injected[...], Autowire(...), Target(...) — follow. A listener class is a singleton: the dispatcher keeps it once built.

Listener methods are inherited: every scanned class carrying one listens, its subclasses included. To keep a base class from listening itself, make it abstract (an ABC with an abstract method) or mark it @exclude. A subscriber base that leaves get_subscribed_events to its subclasses is not registered either.

The container's dispatcher cannot be changed. It is a CompiledEventDispatcher shared for the process's lifetime, so a listener added at runtime would outlive the request, message or task that added it; a listener receiving the dispatcher receives that one too. For listeners of one scope, wrap it: ScopedEventDispatcher(dispatcher).

A bundle tags a service by hand the way the decorator does:

from xtr_event_dispatcher.bundle import LISTENER_TAG, SUBSCRIBER_TAG

services.set(Stock).add_tag(LISTENER_TAG, event=OrderPlaced, method="reserve")
services.set(Audit).add_tag(SUBSCRIBER_TAG, dispatcher="audit")

RegisterListenersPass reads both tags and the decorated functions, and hands every dispatcher its listeners. It runs after every bundle's process hook, so a bundle may tag a listener there too.

A dispatcher is any service tagged DISPATCHER_TAG: the bundle's own, and those another bundle owns. Register one under its name, and every listener naming that dispatcher joins it — it is then injected by name like the configured ones:

from xtr_event_dispatcher.bundle import DISPATCHER_TAG, event_dispatcher_factory

services.set(event_dispatcher_factory("mailer"), qualifier="mailer").add_tag(DISPATCHER_TAG)

The security bundle registers one per firewall this way. The listeners RegisterListenersPass worked out are the dispatcher's ListenerMap; ListenerMap.with_listeners_of(other, events) gives one dispatcher another's listeners of some events, as the security bundle gives each firewall the main dispatcher's security listeners.

Configuration

@configure
def events() -> EventDispatcherConfig:
    return EventDispatcherConfig(
        # a listener of "order.placed" hears OrderPlaced
        event_aliases={"order.placed": OrderPlaced},
        dispatchers=("audit",),  # besides the default one
        trace=None,  # None: trace in debug mode
    )

Another bundle offers names for its events from its prepend_extension hook; an alias the application already maps keeps its event:

builder.prepend_extension_config(
    EventDispatcherConfig, lambda c: c.with_event_aliases({"order.placed": OrderPlaced})
)

In debug mode every dispatcher is wrapped in a TraceableEventDispatcher, reset between units of work (kernel.reset); with the logging bundle active it writes to the "event" channel. A bundle owning a dispatcher traces it the same way with traceable_event_dispatcher_factory.

The bundle's zero-config path builds an empty dispatcher and does no I/O.

debug:event-dispatcher

With a console bundle active, debug mode gets a command listing each event's listeners in the order they run, with their priorities — of any dispatcher, another bundle's included:

app debug:event-dispatcher                      # every event of the default dispatcher
app debug:event-dispatcher OrderPlaced          # the event named, or every event containing it
app debug:event-dispatcher --dispatcher audit   # a named dispatcher
app debug:event-dispatcher --format json        # txt, json, md or xml

A listener service not built yet shows as its LazyListener.

Errors

Every error derives from EventDispatcherError. An exception raised by a listener is not wrapped: it reaches the code that dispatched the event.

Development

Developed in the python-xtr monorepo, under packages/xtr-event-dispatcher; run the commands below from there. The python-xtr-event-dispatcher repository is a read-only copy, so send issues and pull requests to the monorepo.

uv sync
uv run ruff check && uv run ruff format --check && uv run basedpyright && uv run ty check && uv run pytest

License

MIT — see LICENSE.