Event Bus

Async in-process event bus with deterministic ordering, namespace routing, and callback filtering.

What This Package Provides

• EventBus: queue-backed dispatcher.
• Event: immutable envelope delivered to callbacks.
• Subscription: event-type + namespace + filter matching.
• EventBusStateChart: native StateChart bridge for python-statemachine.
• HomeAssistantEventBridge: bidirectional Home Assistant websocket adapter.

Install

Requires Python 3.12 or newer. From a checkout, install the project with uv:

uv add git+https://forgejo.john-stream.com/john/event-bus.git

Quick Start

import asyncio
from typing import TypedDict

from event_bus.bus import EventBus
from event_bus.types import EventType


class CountChangedPayload(TypedDict):
    count: int


type CountEvent = EventType[CountChangedPayload]

COUNT_CHANGED_EVENT_TYPE: CountEvent = EventType("counter.changed")


def print_count_change(event: CountEvent) -> None:
    print(event.event_type, event.namespace, event.data)


async def main() -> None:
    bus = EventBus()
    bus.listen(COUNT_CHANGED_EVENT_TYPE, print_count_change)

    async with bus.run():
        bus.fire(COUNT_CHANGED_EVENT_TYPE, {"count": 123})

        # By default, the async run context will await until the event queue has
        # been fully processed.


asyncio.run(main())

Waiting for One Event

wait_for_event() registers immediately and returns an asyncio.Future that resolves with the first matching Event, not just its payload. Arm the wait before triggering the action that will publish the event:

async with bus.run():
    pending = bus.wait_for_event(COUNT_CHANGED_EVENT_TYPE, count=lambda value: value >= 100)
    try:
        bus.fire(COUNT_CHANGED_EVENT_TYPE, {"count": 123})
        event = await asyncio.wait_for(pending, timeout=5)
        print(event.data)
    finally:
        pending.cancel()

The method requires a running event loop but does not start the bus. It accepts the same namespace, event_filter, wildcard event names, and field filters as listen(). Filter errors are logged and reported to on_error; the wait remains pending for a later match. Independent waits can resolve from the same event.

"First" means delivery order: priority first, then sequence number. Events already queued can satisfy a new wait, but already-delivered events are not replayed. A wait registered during an event's dispatch sees only later events.

Successful delivery removes the temporary subscription immediately. Cancellation removes it on the next event-loop turn; a completed wait skips further filtering even while still present in a dispatch snapshot. Pending waits survive bus shutdown and restart. Keep the Future and cancel it when abandoning a wait, including when the triggering action fails. Timeouts belong to the caller; both asyncio.wait_for() and asyncio.timeout() cancel an unshielded wait on timeout.

Await the Future directly; do not pass it to asyncio.create_task(), which requires a coroutine. Never await a pending event wait inside a bus callback: dispatch awaits callbacks sequentially, so the later event cannot be delivered. Move such workflows to application-owned workers.

Event Names and Payloads

An event has three parts to keep in mind:

Part	Description
Name	Identifies which event happened, such as "user.created".
Payload	The data associated with the specific occurrence, such as a user's ID and email address.
Envelope	The Event object a callback receives. It holds the payload in event.data, along with the event name, namespace, timestamp, and other metadata.

Define the payload in your application, then associate its type with an event name using EventType:

from dataclasses import dataclass

from event_bus.types import EventType


@dataclass(slots=True, frozen=True)
class UserCreatedPayload:
    user_id: str
    email: str


type UserCreatedEventType = EventType[UserCreatedPayload]

USER_CREATED_EVENT: UserCreatedEventType = EventType("user.created")

Here, UserCreatedPayload describes the data, while USER_CREATED_EVENT is the string "user.created" annotated with that payload type.

Register a callback on your bus, then pass the payload to fire(). The bus wraps it in an Event; you do not need to construct the envelope yourself:

def print_created_user(event: UserCreatedEventType) -> None:
    print(event.data.user_id)


bus.listen(USER_CREATED_EVENT, print_created_user)
user = UserCreatedPayload(user_id="42", email="dev@example.com")
bus.fire(USER_CREATED_EVENT, user)

Use stable, descriptive event names in lowercase domain.action form, such as user.created or order.cancelled.

The type annotations help editors and type checkers catch mistakes. They do not validate payloads at runtime.

Lifecycle

• start() returns the queue-processing task. Calling it while a run is active returns the same task. Awaiting it observes shutdown completion and lifecycle failures.
• run() calls start() on entry and stop() on exit. By default (wait=True), it drains queued events and completes their callbacks before shutdown, even when the body raises. With wait=False, exit finishes only the current event and discards the backlog. An explicit shutdown request overrides draining.
• If the run() body raises, its error propagates after successful cleanup. If shutdown also raises, the shutdown error propagates with the body error retained through Python's native exception chaining, rather than an exception group.
• stop() is a no-op before startup. Otherwise it requests shutdown and awaits the runner. The current event's remaining matching subscribers finish normally, including async callbacks; queued events are discarded. Successful completion returns None.
• Shutdown takes effect at event boundaries. A request from a callback, filter, or error hook does not skip the current event's later matching subscribers. An exception stops the callback or filter that raises it, but does not skip later subscribers. No next event starts after the request is observed.
• The shutdown signal is independent of queue capacity. An idle runner wakes immediately when shutdown is requested; shutdown does not need to enqueue an event.
• Shutdown creates no synthetic events. Application resource cleanup belongs in finally blocks or async context managers.
• fire() raises EventBusClosingError while shutdown is pending, including calls from finishing subscribers. After the runner completes, the bus accepts pre-start publishing again.
• stop_event is set when shutdown is requested. It starts clear and remains set after shutdown. Setting it directly requests ordinary shutdown without waiting, including from a subscriber. start() clears it for a new run. Never clear it to resume a stopping run; it does not indicate completed cleanup.
• run_event is set while the runner is active, including while a stopping event finishes, and cleared on exit. Observe it without changing it. It does not indicate application readiness or record earlier runs.
• running is read-only and means run_event is set and stop_event is clear; stopped is its inverse. running can be false while the current event finishes. task_is_active includes scheduled startup and current-event completion. Await the runner or stop() to observe completion.
• queue.join() waits for events to finish their inline callbacks or be discarded. It does not guarantee successful delivery or wait for work handed to external workers.
• Deliberate fatal requests raise EventBusShutdownError after the current event finishes, preserving the reason and original request exception as its cause when present. Infrastructure errors propagate unchanged.
• Completed lifetime tasks are retained until restart. Repeated stop() calls and task awaits return None or re-raise the same exception; cancellation remains cancellation.
• Shutdown never cancels a callback or imposes a deadline. Its latency is the time needed to finish the current event and discard the backlog. Callbacks must be short; long-running work belongs elsewhere.
• Use stop(), stop_event.set(), or request_shutdown() for shutdown. Direct task cancellation follows native asyncio semantics and may interrupt the current event. Cancelling a caller awaiting stop() also cancels the runner. Cancellation-resistant cleanup and cancellation/restart races are not supported lifecycle guarantees.

Downstream Ownership

The application should own bus startup and shutdown, usually with async with bus.run():. Libraries using a shared bus should register subscriptions and manage their own resources with context managers or finally blocks. They should not call stop() or enter run() as independent leases: run() stops the bus on exit even if another caller started it. Registration and publishing do not require waiting for run_event; events can be queued before startup. Bus signals are asyncio.Event control primitives, not dispatched Event messages, so wildcard subscriptions do not receive lifecycle notifications.

To monitor ordinary shutdown during an active run, await stop_event.wait(); this only observes the request. To observe completed shutdown, await the task returned by the owner's start(). Applications that need to isolate that task from cancellation of a monitoring caller can use asyncio.shield(task). Neither signal substitutes for a library's own readiness or teardown protocol.

Deliberate Fatal Shutdown

Ordinary callback exceptions never request bus-wide shutdown. Subscriber code can explicitly use either form:

from event_bus.exceptions import ShutdownRequested


def shutdown_with_api(event):
    bus.request_shutdown("Required service is unavailable")


async def shutdown_with_exception(event):
    raise ShutdownRequested("Required service is unavailable")

Both forms retain the first fatal reason and lead lifecycle callers to receive EventBusShutdownError after the current event's remaining matching subscribers finish. For ordinary shutdown from a subscriber, use bus.stop_event.set() instead. Subscribers must not await stop() or the runner, which would wait for their own completion.

Short Callbacks

Callbacks are short, sequential handoffs to application-owned work. They may enqueue messages, update local state, or schedule work in an application's task group. The bus does not own or await those external workers during shutdown.

There are no callback concurrency limits or execution timeouts. Sync callbacks, filters, error hooks, and log handlers must not block the event loop. An async callback that waits indefinitely also prevents further delivery and shutdown.

Callbacks must not wait for a later bus event, another subscriber, queue.join(), or bus shutdown: sequential delivery cannot make progress until the current callback returns. Move dependency-bearing workflows to application-owned workers.

Event Envelope

Each queued event contains:

• event_type: the event name, represented by an EventType[Payload] string.
• data: the application payload. Direct Event construction defaults to None; fire() requires an explicit payload argument, which may be None.
• origin
• context
• time_fired (UTC)
• namespace
• suppressed: whether delivery requires receive_suppressed=True (default False).
• priority
• seq: the sequence number used to order events with equal priority.

API Reference

listen(...)

def listen(
    event_type,
    callback,
    *,
    receive_suppressed=False,
    event_filter=None,
    namespace=None,
    **field_filters,
) -> None:
    ...

• Registers a callback for one namespace (bus.default_namespace when omitted).
• callback can be sync or async and always receives an Event[...] envelope. It must return None or an awaitable resolving to None.
• event_filter runs first and receives the full envelope.
• field_filters are evaluated against event.data.
  • If event.data is a mapping, keys are used.
  • Otherwise, attribute lookup is used.

fire(...)

def fire(
    event_type,
    event_data,
    origin=None,
    context=None,
    *,
    suppressed=False,
    namespace=None,
    priority=None,
    seq=None,
) -> None:
    ...

• event_type selects the event name; event_data supplies its payload.
• Immediately enqueues an event without waiting for callbacks.
• Defaults:
  • namespace=bus.default_namespace (initially "default")
  • priority=100
    • seq (sequence number) auto-increments per bus instance. An explicit seq does not advance the counter.

Dispatch Semantics

• Queue is asyncio.PriorityQueue.
• Lower numeric priority is dispatched first.
• For equal priority, lower seq is dispatched first (FIFO for same priority).
• Routing is namespace-scoped: listeners only receive events from their registered namespace.
• suppressed=True skips ordinary subscribers and reaches only subscriptions with receive_suppressed=True. Events are always queued; suppression does not change dispatch timing.
• allocate_sequence() explicitly advances the sequence counter.

Matching and Filters

Event type matching

• Exact event type matches are supported.
• Wildcards are supported only when * appears in the subscription event type.
  • Example: custom.* matches custom.created.
• ? is treated as a literal character for deciding wildcard mode.

Field filters

field_filters values must be either:

• str: exact match or glob pattern (fnmatchcase semantics).
• callable: predicate receives field value and returns bool.

Any other filter value type raises ValueError.

Missing fields fail string filters. Callable filters receive an opaque sentinel distinct from None and decide whether to accept it. Predicate and payload-access failures are reported for that subscriber; other subscribers and later events continue. Direct calls to Subscription.matches() or check_filters() still propagate errors; containment belongs to bus dispatch.

Callback Execution and Errors

• Sync callbacks run inline in the dispatcher loop and must be nonblocking.
• Returned awaitables, including futures, are awaited inline. Subscribers run in registration order, and an event finishes all its callbacks before the next event starts.
• A failure in one subscriber's matching or callback code is reported and does not prevent other subscribers or subsequent events from being processed.
• Reports preserve the original exception and traceback, subscription, event, and phase: matching, invocation, or await. No automatic retries or rollback are performed.
• Failures are logged once. An optional synchronous on_error hook receives SubscriberFailure; hook failures are logged without recursively invoking it. The hook must not block or return an awaitable.
• Shutdown requests are control flow, not subscriber failures. Unexpected bus infrastructure failures remain visible to lifecycle callers.
• The event envelope is frozen, but its payload may be mutable and shared between subscribers. This is fault containment, not a security sandbox or successful-delivery guarantee.

Extensions

State Chart

Use EventBusStateChart as your base class when you want state transitions from python-statemachine to be emitted on the event bus. It subclasses StateChart directly and uses its native processing and error-handling defaults.

Basic usage:

import asyncio

from statemachine import State

from event_bus.bus import EventBus
from event_bus.extensions.statemachine import EVENT_STATEMACHINE_CHANGE
from event_bus.extensions.statemachine import EventBusStateChart
from event_bus.extensions.statemachine import StateMachineEvent


class DoorStateMachine(EventBusStateChart):
    closed = State(initial=True)
    open = State()

    open_door = closed.to(open)
    close_door = open.to(closed)


def print_transition(event: StateMachineEvent) -> None:
    print(event.event_type, event.data.source, event.data.target)


async def main() -> None:
    bus = EventBus()
    door_machine = DoorStateMachine(bus=bus)

    bus.listen(EVENT_STATEMACHINE_CHANGE, print_transition)

    async with bus.run():
        door_machine.open_door()
        door_machine.close_door()

        # Process all outstanding items in the event bus queue
        await bus.queue.join()


asyncio.run(main())

Transitions are published under the event-name constant EVENT_STATEMACHINE_CHANGE. Callbacks receive a StateMachineEvent, an alias for the Event[EventData] envelope. Its .data is the transition payload supplied by python-statemachine, including the source state, target state, and trigger information.

Transition Hooks

from statemachine import State
from statemachine.event_data import EventData

from event_bus.extensions.statemachine import EventBusStateChart


class DoorStateMachine(EventBusStateChart):
    closed = State(initial=True)
    open = State()

    open_door = closed.to(open)
    close_door = open.to(closed)

    def on_open_door(self, event_data: EventData) -> None:
        print(event_data.state)

These hooks come from python-statemachine itself and run as part of transition processing. Bus callbacks run separately when the queued transition event is dispatched.

Listen for Target States

Use listen_state() when you want a callback only after a specific target state is reached.

import asyncio

from statemachine import State

from event_bus.bus import EventBus
from event_bus.extensions.statemachine import EventBusStateChart
from event_bus.extensions.statemachine import StateMachineEvent


class DoorStateMachine(EventBusStateChart):
    closed = State(initial=True)
    open = State()

    open_door = closed.to(open)
    close_door = open.to(closed)


def print_door_opened(event: StateMachineEvent) -> None:
    print("Reached:", event.data.state)


async def main() -> None:
    bus = EventBus()
    door_machine = DoorStateMachine(bus=bus)

    door_machine.listen_state(door_machine.open, print_door_opened)

    async with bus.run():
        door_machine.open_door()
        await bus.queue.join()


asyncio.run(main())

listen_state() checks the publishing machine's identity and the transition's target field. It receives only this instance's transitions into the state you passed in.

Listen for Transition Triggers

Use listen_event() to receive bus envelopes for transitions caused by a specific state-machine trigger, regardless of the target state. A trigger such as open_door causes a transition; it is distinct from the bus event name EVENT_STATEMACHINE_CHANGE.

import asyncio

from statemachine import State

from event_bus.bus import EventBus
from event_bus.extensions.statemachine import EventBusStateChart
from event_bus.extensions.statemachine import StateMachineEvent


class DoorStateMachine(EventBusStateChart):
    closed = State(initial=True)
    open = State()

    open_door = closed.to(open)
    close_door = open.to(closed)


def print_transition_trigger(event: StateMachineEvent) -> None:
    print("Trigger:", event.data.trigger_data.event)


async def main() -> None:
    bus = EventBus()
    door_machine = DoorStateMachine(bus=bus)

    door_machine.listen_event(door_machine.open_door, print_transition_trigger)

    async with bus.run():
        door_machine.open_door()
        door_machine.close_door()
        await bus.queue.join()


asyncio.run(main())

listen_event() checks the publishing machine's identity and trigger_data.event. Callbacks receive bus events only for this instance's transitions using the specified trigger.

Home Assistant

Use HomeAssistantEventBridge as a thin bidirectional adapter between Home Assistant events and this package's EventBus.

Events are bridged as follows:

• Home Assistant -> EventBus: HA websocket events are published into a configurable EventBus namespace.
• EventBus -> Home Assistant: EventBus events from that namespace are fired back into Home Assistant.

The bridge uses homeassistant_api.AsyncWebsocketClient and keeps payloads intact when event data is a mapping.

run() owns both connections and forwarding tasks. Worker failures cancel sibling work and propagate as an ExceptionGroup; running reflects worker health. Context exit cancels forwarding rather than draining queued outbound events.

Basic usage:

import asyncio

from event_bus.bus import EventBus
from event_bus.extensions.hass import HomeAssistantEventBridge
from event_bus.types import EventType


DOOR_OPENED_EVENT_TYPE: EventType[dict[str, str]] = EventType("door.opened")


async def main() -> None:
    bus = EventBus()

    bridge = HomeAssistantEventBridge(
        bus=bus,
        ws_url="ws://homeassistant.local:8123/api/websocket",
        access_token="YOUR_LONG_LIVED_ACCESS_TOKEN",
        namespace="home_assistant",
    )

    async with bus.run(), bridge.run():
        # This event is forwarded to Home Assistant as a fired event.
        bus.fire(
            event_type=DOOR_OPENED_EVENT_TYPE,
            event_data={"entity_id": "binary_sensor.front_door"},
            namespace="home_assistant",
        )

        await asyncio.sleep(5)


asyncio.run(main())

Notes:

• Use the same namespace for both directions so HA-origin events and outbound bus events stay grouped.
• The bridge namespace defaults to "default", independently of bus.default_namespace.
• HA-origin events are tagged with origin="homeassistant_api_bridge" to prevent echo loops.
• incoming_event_type optionally restricts inbound Home Assistant events; it does not restrict outbound forwarding.
• Suppressed bus events are forwarded too. Suppression is not a privacy boundary for the bridge.
• A None payload sends no event data. Other non-mapping payloads are sent to Home Assistant as value=<payload>.
• The application owns the bus lifecycle separately; bridge.run() does not start or stop the bus.

Testing

Install test dependencies and run the unit tests (the live Home Assistant integration tests require a configured server):

uv sync --group test
uv run pytest tests/test_event_bus.py tests/test_subscription.py tests/test_statemachine.py tests/test_hass_bridge.py

To include live integration tests, set HASS_URI (the websocket URL) and TOKEN in the environment or a local .env file, then run uv run pytest. Without those values, live tests are skipped.

Test coverage for the package with missing-line reporting and XML output:

uv run pytest tests/test_event_bus.py tests/test_subscription.py tests/test_statemachine.py tests/test_hass_bridge.py --cov=event_bus --cov-report=term-missing --cov-report=xml