emitix

Lightweight event bus for Python applications. No dependencies, fully typed, works with both sync and async code.

pip install emitix

Quick start

import emitix

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

emitix.emit("user.created", "Ada")
# Welcome, Ada!

The module-level functions use one shared bus. Create your own when you want separate buses:

from emitix import EventBus

bus = EventBus()
bus.on("order.paid", lambda order_id: print("paid", order_id))
bus.emit("order.paid", 42)

Features

Any arguments

Whatever you pass to emit is passed to the handlers, and emit returns their results.

bus.on("add", lambda a, b: a + b)
bus.emit("add", 2, 3)  # [5]

Wildcards

Subscriptions accept shell-style patterns (*, ?, [abc]). Use pass_event=True to receive the event name as the first argument.

@bus.on("user.*", pass_event=True)
def audit(event, user):
    print(event, user)

bus.emit("user.created", "Ada")  # user.created Ada
bus.emit("user.deleted", "Ada")  # user.deleted Ada

One-shot handlers

@bus.once("app.ready")
def warm_cache():
    ...

Priorities

Higher priority runs first. Handlers with the same priority run in the order they were registered.

bus.on("save", validate, priority=10)
bus.on("save", write_to_disk)

Async handlers

emit_async awaits async def handlers and calls regular ones normally. Handlers run one after another, in priority order.

@bus.on("job.done")
async def notify(job_id):
    await send_notification(job_id)

await bus.emit_async("job.done", 7)

Calling plain emit for an event that has an async handler raises TypeError.

Unsubscribing

bus.off("save", write_to_disk)  # remove one handler
bus.off("save")                 # remove every handler for "save"
bus.clear()                     # remove everything

API

An exception raised by a handler propagates to the caller of emit and stops the remaining handlers. All methods are thread-safe.

Development

pip install -e ".[dev]"
pytest

License

MIT