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

Contents

• Why an event bus?
• Quick start
• Core concepts
• Use cases
• Patterns and ordering
• Error handling
• API reference
• Behaviour and limits
• Examples
• Migrating from 0.1.1
• Development

---

Why an event bus?

Without a bus, the code that knows what happened has to know who cares:

def register_user(email):
    user_id = save_user(email)      # core
    send_welcome_email(user_id)     # marketing
    create_default_settings(user_id)  # settings
    index_for_search(user_id)       # search

Every new reaction edits register_user, imports another module, and grows the test setup. With a bus, the core publishes one fact and stays closed for modification:

def register_user(bus, email):
    user_id = save_user(email)
    bus.emit("user.created", user_id)
    return user_id

Subscribers live wherever they belong:

@bus.on("user.created")
def send_welcome_email(user_id): ...

@bus.on("user.created")
def create_default_settings(user_id): ...

@bus.on("user.created")
def index_for_search(user_id): ...

The trade-off is intentional: an event bus decouples producers from consumers, which makes flows harder to follow in exchange for making them easy to extend. It shines in apps with cross-cutting reactions, plugin-style extension points, and tests that need to observe behaviour without mocks.

Quick start

from relaykit_tools import EventBus

bus = EventBus()

# A handler can be any callable.
def greet(name):
    print(f"Hello, {name}!")

bus.on("user.joined", greet)
bus.emit("user.joined", "Alex")
# Hello, Alex!

emit returns how many handlers ran, which makes "did anything listen?" easy to check:

print(bus.emit("user.joined", "Sam"))  # 1
print(bus.emit("user.left", "Sam"))    # 0

Subscribe with a decorator

@bus.on("order.created")
def process_order(order_id):
    print(f"Processing order {order_id}")

bus.emit("order.created", 123)

Run a handler once

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

bus.emit("app.started")   # Caches warmed
bus.emit("app.started")   # nothing happens

Watch everything

A wildcard handler receives the event name as its first argument:

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

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

Unsubscribe

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

---

Core concepts

Concept	What it is
Event	A string name plus the arguments it was published with, e.g. "order.paid". Names are case-sensitive and 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 that matches several event names: "*" or "user.*".
Scope	bus.scope("user") — a view that prefixes every event with a namespace and can remove everything it registered in one call.
Recorder	EventRecorder(bus) — records events so tests can assert on them.

Use cases

1. Decouple modules inside one app

A checkout service publishes facts; invoicing, email and analytics react without the checkout code importing any of them.

# checkout.py
def checkout(bus, cart, user):
    order_id = create_order(cart, user)
    bus.emit("order.paid", order_id=order_id, total=cart.total, user_id=user.id)
    return order_id

# billing.py
@bus.on("order.paid")
def create_invoice(order_id, total, user_id):
    invoices.create(order_id=order_id, amount=total, customer=user_id)

# analytics.py
@bus.on("order.paid")
def track_purchase(order_id, total, user_id):
    analytics.track("purchase", {"order": order_id, "value": total})

Wiring happens once, usually where the bus is created:

bus = EventBus()
import billing, analytics   # each module subscribes at import time

Prefer explicit registration in one place (wiring.py) if import side effects make you uneasy — both work.

2. Plugin systems and ordered hooks

Higher-priority handlers run first, which is how you put auditing or validation in front of "normal" work:

@bus.on("*", priority=100)
def audit(event_name, *args, **kwargs):
    audit_log.write(event_name, args, kwargs)

@bus.on("document.save", priority=10)
def validate(document):
    if not document.is_valid():
        raise ValueError("invalid document")

@bus.on("document.save")
def save(document):
    documents.save(document)

document.save now runs audit → validate → save, and the audit log also sees every other event on the bus.

3. Namespaces and request-scoped listeners

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 can remove everything it registered, which is ideal for something with a lifetime — a request, a session, a plugin you can install and uninstall:

def handle_request(bus, request):
    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:

users.on("*", lambda name, *args: print("user event:", name))
users.emit("deleted", 7)   # user event: user.deleted

Nested scopes compose: bus.scope("order").scope("checkout") is "order.checkout", and closing the parent closes the children too.

4. Async handlers

Register coroutine functions normally; 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 can be mixed in
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 one starts. If you call plain emit() on an event with an async handler, RelayKit raises a TypeError that tells you to use emit_async() — it never silently drops a coroutine.

5. Wait for something to happen

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

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

wait_for also accepts patterns and a predicate:

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

It unsubscribes itself afterwards, including when it times out (asyncio.TimeoutError).

6. Test what a function publishes

The recorder removes the need to mock collaborators that would have been called directly:

from relaykit_tools import EventBus, EventRecorder

def test_register_user_emits_events():
    bus = EventBus()
    recorder = EventRecorder(bus)

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

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

Record only one namespace with EventRecorder(bus, "email.*"), cap memory with limit=, or drive it by hand with start() / stop().

7. Observe without touching the code

A single wildcard listener gives you logging, metrics or tracing for free:

@bus.on("*")
def metrics(event_name, *args, **kwargs):
    statsd.incr(f"events.{event_name}")

Because subscribers are just handlers, you can add one in a test, in a debug console, or behind a feature flag.

---

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

Wildcard handlers receive the event name first, before the emitted arguments:

@bus.on("user.*")
def handle(event_name, *args, **kwargs):
    ...
bus.emit("user.created", 42)      # handle("user.created", 42)

Handlers added while an event is being dispatched only run on the next emit, and once handlers are removed before they are called. Dispatch always works on a snapshot, so unsubscribing inside a handler is safe.

Error handling

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

try:
    bus.emit("task.started")
except Exception as exc:
    print(f"Event handler failed: {exc}")

Three ways to handle failures:

# 1. Log and keep going (errors are reported, dispatch is not interrupted).
bus = EventBus(on_error=lambda exc, event: logger.exception("event %s", event.name))

# 2. Inspect failures at the call site.
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

# 3. Handle them in the handler.
@bus.on("task.started")
def start():
    try:
        ...
    except Exception:
        logger.exception("start failed")

on_error and emit_collect combine: with on_error set, the callback is invoked and dispatch continues; emit_collect still reports the errors.

API reference

Everything below is importable from the package root.

EventBus

from relaykit_tools import 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, so it can decorate.
once(event, handler=None, *, priority=0)	Same, but the handler is removed before its first call.
off(event, handler=None) -> bool	Remove one handler, or all handlers for that event name/pattern. True if something was removed.
clear(event=None)	Remove listeners for one event, or all listeners.
emit(event, *args, **kwargs) -> int	Publish; returns the number of handlers invoked.
emit_collect(event, *args, **kwargs) -> DispatchResult	Publish, collect results and errors instead of raising.
await emit_async(event, *args, **kwargs) -> int	Publish and await coroutine handlers.
await emit_async_collect(event, *args, **kwargs) -> 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 (including patterns).
listeners(event) -> tuple[Handler, ...]	The handlers that would run, in dispatch order.
events() -> tuple[str, ...]	Exact event names with listeners.
patterns() -> tuple[str, ...]	Wildcard patterns with listeners.
scope(prefix) -> ScopedBus	Namespaced view.

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

Attribute	Description
event	The emitted Event.
invoked	Number of handlers called.
results	Return values of handlers that succeeded, in order.
errors	Exceptions raised by handlers.
ok	True when no handler raised.

ScopedBus

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

scope.off("event") without a handler removes only the listeners that scope registered — it never touches listeners added directly on the bus. Once closed, a scope rejects new subscriptions with RuntimeError.

EventRecorder

recorder = EventRecorder(bus, "user.*", limit=100)
recorder.start()            # also works 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 it was emitted; returns the last one.
assert_not_emitted(event)	Raise AssertionError if it was emitted.

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 itself does not marshal work onto other threads.
• Async handlers are awaited sequentially (no gather), which keeps ordering predictable. Use your own task group inside a handler if you want concurrency.
• Event names are case-sensitive and free-form strings (dots recommended).
• No serialization, retries, persistence or delivery guarantees. This is an in-process bus, not a message broker.
• No handler 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 use scope.close() or off() to release them.

Examples

Runnable scripts live in examples/:

File	Shows
quickstart.py	Publishing, decorators, one-shot handlers, wildcards.
plugin_system.py	Priorities, namespace patterns, installable plugins, clean teardown.
async_pipeline.py	emit_async, wait_for, predicates, error propagation.
testing_events.py	Asserting on emitted events with EventRecorder.

From a checkout (no install needed):

PYTHONPATH=src python examples/quickstart.py

Migrating from 0.1.1

The public import path is unchanged; from relaykit_tools import EventBus still works. Behavioural changes and additions in 0.1.2:

• once handlers are no longer wrapped, so off(event, handler) works with the function you registered, and listener_count is accurate inside handlers.
• listener_count(event) now counts handlers that would actually run for that event, including matching wildcard listeners.
• New: priority ordering, namespace patterns ("user.*"), emit_collect, emit_async, wait_for, ScopedBus, EventRecorder, Event, DispatchResult.
• EventBus.emit still returns an int and still raises handler exceptions by default.

See CHANGELOG.md for the full list.

Development

git clone <your-fork>
cd relaykit
python -m venv .venv
.venv/bin/pip install -e ".[dev]"     # Windows: .venv\Scripts\pip
.venv/bin/python -m pytest

Build and check a release:

python -m build
python -m twine check dist/*

Requirements

• Python 3.10 or newer.
• No runtime dependencies.

License

MIT — see LICENSE.