RelayKit Tools

A tiny, dependency-free event bus for Python.

RelayKit lets the parts of an application talk to each other through events instead of direct calls. A module publishes what happened; anyone who cares subscribes. Neither side needs to import the other, so features can be added, removed or tested in isolation.

from relaykit_tools import EventBus

bus = EventBus()

@bus.on("user.created")
def send_welcome_email(user_id):
    print(f"Welcome, user {user_id}!")

bus.emit("user.created", 42)   # Welcome, user 42!

• No dependencies, no framework, no configuration.
• Sync and async handlers, ordered by priority.
• Wildcard patterns ("*", "user.*") and namespaced scopes.
• Built-in recorder for asserting on events in tests.
• Under 1,000 lines of type-annotated Python, Python 3.10+.

Installation

pip install relaykit-tools

Quick start

from relaykit_tools import EventBus

bus = EventBus()

def greet(name):
    print(f"Hello, {name}!")

bus.on("user.joined", greet)        # or decorate: @bus.on("user.joined")
bus.emit("user.joined", "Alex")     # Hello, Alex!

print(bus.emit("user.joined", "Sam"))   # 1 — emit returns the handler count
print(bus.emit("user.left", "Sam"))     # 0 — nobody is listening

bus.off("user.joined", greet)   # remove one handler
bus.off("user.joined")          # remove every handler for that event
bus.clear()                     # remove everything on the bus

on, once and off work as methods or decorators, and every emit variant returns either the number of handlers that ran or a DispatchResult.

One-shot and wildcard listeners

@bus.once("app.started")
def warm_caches():
    print("Caches warmed")

@bus.on("*")
def log_everything(event_name, *args, **kwargs):
    print(event_name, args, kwargs)

bus.emit("app.started")     # Caches warmed, then app.started () {}
bus.emit("user.created", 42)  # user.created (42,) {}

A wildcard listener always receives the concrete event name as its first argument, before the arguments that were emitted.

Core concepts

Concept	What it is
Event	A name plus its arguments, e.g. "order.paid". Case-sensitive; dotted names are the convention.
Handler	Any callable — def, lambda, method, callable object. Async functions are supported.
Bus	EventBus, the registry of handlers and the dispatcher. Most apps create one and share it.
Pattern	A subscription matching several names: "*" or "user.*".
Scope	bus.scope("user"), a view that prefixes every event with a namespace and can be torn down at once.
Recorder	EventRecorder(bus), which records events so tests can assert on them.

Namespaced scopes

bus.scope("user") prefixes registration and emission, so a module can say "created" instead of repeating "user.created":

users = bus.scope("user")

@users.on("created")
def send_welcome(user_id): ...

users.emit("created", 42)     # publishes "user.created"

A scope removes everything it registered in one call, which suits anything with a lifetime — a request, a session, an installable plugin:

with bus.scope(f"request.{request.id}") as scope:
    scope.on("files.changed", request.refresh)
    bus.emit(f"request.{request.id}.files.changed")
# listeners are gone here

Subscribing with "*" inside a scope listens to that namespace only, nested scopes compose (bus.scope("order").scope("checkout") is "order.checkout"), and closing a parent closes its children. A closed scope refuses new subscriptions with RuntimeError. scope.off("event") only ever removes the listeners that scope registered.

Async handlers

Register coroutine functions normally and await them with emit_async:

@bus.on("upload.finished")
async def resize(upload_id):
    await images.resize(upload_id)

@bus.on("upload.finished")          # sync handlers mix in freely
def index(upload_id):
    search.index(upload_id)

await bus.emit_async("upload.finished", 3)

Handlers run in order and each async handler is awaited before the next starts. Calling plain emit() on an event with an async handler raises a TypeError that points at emit_async() — a coroutine is never silently dropped.

Waiting for an event

wait_for returns the next matching Event, which is handy for startup handshakes and integration tests:

event = await bus.wait_for("app.ready", timeout=5)
print(event.name, event.kwargs)      # app.ready {'port': 8080}

failure = await bus.wait_for(
    "health.check",
    timeout=30,
    predicate=lambda event: not event.kwargs["ok"],
)

It accepts patterns and a predicate, removes its listener afterwards, and raises asyncio.TimeoutError when nothing acceptable arrives in time.

Error handling

By default a handler exception propagates to whoever emitted the event and the remaining handlers do not run.

bus = EventBus(on_error=lambda exc, event: logger.exception("event %s", event.name))

result = bus.emit_collect("task.started")
result.invoked    # handlers called
result.results    # return values of the handlers that succeeded
result.errors     # exceptions raised
result.ok         # True when errors is empty

on_error logs a failure and continues dispatching; emit_collect and emit_async_collect report failures instead of raising. The two combine.

Patterns and ordering

Subscription	Runs for
"user.created"	exactly "user.created"
"user.*"	"user" and anything below it — "user.created", "user.profile.updated"
"*"	every event

Matching is case-sensitive and prefix based: "user.*" does not match "userlike.created" or "admin.user.created".

Dispatch order is deterministic: highest priority first, then registration order. Priority defaults to 0 and may be negative.

bus.on("job.run", audit, priority=10)
bus.on("job.run", work)
bus.on("job.run", cleanup, priority=-10)
# audit → work → cleanup

Handlers added during dispatch only run on the next emit, and one-shot handlers are removed before they are called — but only once they are actually reached, so an exception earlier in the dispatch does not consume them. Dispatch works on a snapshot, so unsubscribing inside a handler is safe.

API reference

Everything below is importable from the package root.

EventBus

bus = EventBus()                    # or EventBus(on_error=callback)

Method	Description
on(event, handler=None, *, priority=0)	Subscribe; usable as on(event, fn) or @on(event). Returns the handler.
once(event, handler=None, *, priority=0)	Same, but removed before its first call.
off(event, handler=None) -> bool	Remove one handler, or all handlers for an event name/pattern.
clear(event=None)	Remove listeners for one event, or all of them.
emit(event, *args, **kwargs) -> int	Publish; returns the number of handlers invoked.
emit_collect(event, *args, **kwargs) -> DispatchResult	Publish, collecting results and errors instead of raising.
await emit_async(event, *args, **kwargs) -> int	Publish, awaiting coroutine handlers.
await emit_async_collect(...) -> DispatchResult	Async collecting variant.
await wait_for(event, *, timeout=None, predicate=None) -> Event	Wait for the next matching event.
listener_count(event=None) -> int	Listeners in total, or all that would run for event (patterns included).
listeners(event) -> tuple[Handler, ...]	The handlers that would run, in dispatch order.
events() -> tuple[str, ...] / patterns() -> tuple[str, ...]	Exact names / wildcard patterns with listeners.
scope(prefix) -> ScopedBus	Namespaced view.

ScopedBus

Created with bus.scope(prefix). Supports on, once, off, emit, emit_collect, emit_async, emit_async_collect, wait_for, nested scope(), close() and the with statement. Properties: bus, prefix, closed.

Event

Returned by wait_for and stored by EventRecorder.

event.name       # "order.paid"
event.args       # (123,)
event.kwargs     # {"total": 99.0}
event.payload    # 123 — first positional argument, or None
str(event)       # "order.paid(123, total=99.0)"

DispatchResult

event, invoked, results, errors, and ok (true when nothing raised).

EventRecorder

recorder = EventRecorder(bus, "user.*", limit=100)
recorder.start()            # also usable as a context manager
recorder.stop()

Member	Description
events	Recorded Event objects, oldest first.
names()	Recorded event names.
count(event), last(event=None)	Occurrence count / most recent event.
clear()	Forget recorded events, keep listening.
assert_emitted(event, times=None) -> Event	Raise AssertionError unless emitted; returns the last one.
assert_not_emitted(event)	Raise AssertionError if it was emitted.

bus = EventBus()
recorder = EventRecorder(bus)

with recorder:
    register_user(bus, "alex@example.com")

assert recorder.names() == ["user.registered", "email.welcome_requested"]
assert recorder.assert_emitted("user.registered").kwargs["email"] == "alex@example.com"
recorder.assert_not_emitted("email.bounced")

Pytest fixtures

Installing the package registers a pytest plugin with two fixtures: bus, a fresh EventBus per test, and recorder, an EventRecorder already listening on it.

def test_register_user_emits_events(bus, recorder):
    register_user(bus, "alex@example.com")

    recorder.assert_emitted("user.registered")
    recorder.assert_not_emitted("email.bounced")

Define your own bus or recorder fixture to override them.

Behaviour and limits

• Synchronous by default. emit blocks until every handler returns; async handlers need emit_async.
• Thread-safe registration and dispatch via a re-entrant lock, so handlers may subscribe, unsubscribe or emit from inside a handler. The lock is not held while handlers run, so a handler may block; the bus does not marshal work onto other threads.
• Async handlers are awaited sequentially (no gather), keeping ordering predictable. Use a task group inside a handler for concurrency.
• Event names are free-form strings (dots recommended) and case-sensitive.
• No serialization, retries, persistence or delivery guarantees. This is an in-process bus, not a message broker.
• No return-value aggregation on emit; use emit_collect when you need the results.
• Callbacks are held strongly while subscribed. Long-lived buses that subscribe short-lived objects (requests, sessions) should call off() or scope.close() to release them.

Requirements

• Python 3.10 or newer.
• No runtime dependencies.

License

MIT. Release notes live in CHANGELOG.md in the source repository.