dispatchbus 0.1.0a1

Dispatchbus Python project.

dispatchbus

dispatchbus is an in-memory Python message bus for applications that want explicit command and event dispatch without bringing in a framework.

Alpha release: dispatchbus is in early prerelease status. Expect rough edges and breaking changes before a stable release.

It supports:

• one command handler per command
• zero or more event handlers per event
• async-first APIs with synchronous bridge methods
• middleware around command and event dispatch
• follow-up events emitted from handlers via a context object
• observability subscribers for dispatch and handler lifecycle events
• optional debug subscribers for human-friendly or key/value logging
• sequential or concurrent event handler execution

Requirements

• Python 3.12+

Installation

With uv:

uv add dispatchbus

With pip:

pip install dispatchbus

To use the built-in SQLite outbox implementation, install the optional sqlite extra:

pip install 'dispatchbus[sqlite]'

Without that extra, dispatchbus.outbox still provides the outbox protocols and helpers for custom implementations, but SQLiteOutboxStorage will raise an ImportError with install instructions if accessed.

Quick start

from dataclasses import dataclass

from dispatchbus import CommandBase, EventBase, MessageBus


@dataclass(frozen=True)
class CreateUser(CommandBase):
    message_name = "user.create"
    name: str


@dataclass(frozen=True)
class UserCreated(EventBase):
    message_name = "user.created"
    user_id: int


async def create_user(command: CreateUser, context) -> str:
    user_id = len(command.name)
    context.emit(UserCreated(user_id=user_id))
    return command.name.upper()


async def on_user_created(event: UserCreated) -> None:
    print(f"user created: {event.user_id}")


bus = MessageBus()
bus.register_command_handler(CreateUser, create_user)
bus.register_event_handler(UserCreated, on_user_created)

result = await bus.send(CreateUser(name="ada"))
print(result)  # ADA

dispatchbus does not depend on Pydantic. Dataclasses, Pydantic models, attrs classes, and similar payload types can all be used as long as they subclass CommandBase or EventBase.

Core concepts

Commands

Commands are sent with await bus.send(command).

Commands must be subclasses of CommandBase. Root commands are wrapped and stamped automatically when they enter the bus.

• A command must have exactly one registered handler.
• Registering a second command handler for the same message type raises DuplicateCommandHandlerError.
• Sending a command with no handler raises NoCommandHandlerError.

Events

Events are published with await bus.publish(event).

Events must be subclasses of EventBase. Root events are wrapped and stamped automatically when they enter the bus. Follow-up events emitted through context.emit(...) inherit correlation and causation automatically.

• An event may have zero, one, or many handlers.
• Publishing an event with no handlers is allowed.
• If one or more event handlers fail, EventPublicationError is raised and exposes the collected failures.

Advanced metadata

For normal application code, use plain payload models and let dispatchbus manage metadata at runtime.

• root commands and events are stamped automatically
• follow-up events inherit correlation and causation automatically
• pre-stamped messages are preserved for compatibility and adapter scenarios
• advanced code can read metadata with get_metadata(message)
• MessageMetadata, new_root_metadata(), and derive_child_metadata() remain available for explicit integrations

get_metadata(CreateUser(name="ada")) raises for a plain payload object unless you are holding a RuntimeMessage or using a legacy/pre-stamped message object. Dispatching a plain payload does not make that original payload object metadata-readable later.

Handler lookup

Handler lookup uses the exact runtime type of the message.

If you register a handler for BaseEvent, it will not automatically receive DerivedEvent instances unless you also register DerivedEvent explicitly.

Handler shapes

Handlers and subscribers may be:

• async def callables, or
• synchronous callables that return a final value immediately.

A synchronous callable that returns a coroutine or other awaitable is rejected at runtime.

Handlers can optionally accept a context argument:

async def handle(message, context) -> None: ...


async def handle(message, *, context) -> None: ...

The context lets a handler emit follow-up events:

context.emit(SomeEvent(...))

Event concurrency

Configure event execution with event_concurrency:

bus = MessageBus(event_concurrency="sequential")

Supported modes:

• "concurrent" (default): event handlers start together and may finish in any order
• "sequential": event handlers run in registration order

Semantics:

• command dispatch always targets a single handler
• sequential event mode preserves handler execution order for a given event
• concurrent event mode does not guarantee completion order
• follow-up events emitted by concurrent handlers may interleave

Middleware

Middleware wraps dispatch and can intercept both send() and publish() pipelines.

from typing import Any


async def logging_middleware(message: Any, next_call):
    print(f"before {type(message).__name__}")
    try:
        return await next_call(message)
    finally:
        print(f"after {type(message).__name__}")


bus = MessageBus(middleware=[logging_middleware])

Middleware receives the message and a next_call awaitable callback.

Observability subscribers

Subscribers can observe dispatch lifecycle events emitted by the bus:

• DispatchStarted
• DispatchFinished
• HandlerStarted
• HandlerFinished
• HandlerFailed

from dispatchbus import DispatchFinished, DispatchStarted, HandlerFailed, MessageBus


async def audit(event: object) -> None:
    match event:
        case DispatchStarted(operation=operation, message_type=message_type):
            print(f"starting {operation} for {message_type.__name__}")
        case DispatchFinished(operation=operation, success=success, duration_ms=duration_ms):
            print(f"finished {operation}: success={success} duration_ms={duration_ms:.2f}")
        case HandlerFailed(handler_name=handler_name, error=error):
            print(f"handler failed: {handler_name}: {error}")


bus = MessageBus(subscribers=[audit])

Subscribers are isolated from dispatch: subscriber exceptions are ignored.

Debug subscribers

dispatchbus.debug includes ready-made subscribers for development-time logging:

• DebugSubscriber.human(...)
• DebugSubscriber.key_value(...)
• debug_subscriber_human(...)
• debug_subscriber_key_value(...)

from dispatchbus import MessageBus
from dispatchbus.debug import DebugSubscriber


bus = MessageBus(
    subscribers=[DebugSubscriber.human(include_dispatch=True)],
)

These subscribers can write to:

• a text stream
• a logging.Logger
• a Rich console if rich is installed

Async and sync APIs

Use the async methods from async code:

• await bus.send(...)
• await bus.publish(...)
• await bus.aclose()

Use the sync bridge methods from synchronous code:

• bus.send_sync(...)
• bus.publish_sync(...)
• bus.close()

Calling sync bridge methods inside an active event loop raises BusUsageError.

Closing behavior

When closing begins, the bus stops accepting new top-level sends and publishes. Already-running dispatch can still finish, including nested event publication triggered during that work.

Outbox support

dispatchbus.outbox includes protocol-based outbox building blocks:

• OutboxMessage
• OutboxRetentionPolicy
• OutboxStorage
• MessageSerializer
• EventPublisher
• JSONSerializer
• OutboxProcessor
• OutboxWorker

If you install the optional sqlite extra, it also exposes:

• SQLiteOutboxStorage

SQLite outbox processing uses claim-based polling with at-least-once delivery semantics:

• workers claim unpublished rows before publishing them
• successful publishes set published_at and clear claimed_at
• failed publishes release claims for retry
• stale claims can be reclaimed after the configured timeout
• duplicate pickup across workers is reduced, but exactly-once delivery is not guaranteed

Published-row eviction is available as an explicit storage API and optional worker automation:

• eviction only applies to rows where published_at is not null
• worker-driven eviction is opt-in and disabled by default
• when enabled, eviction deletes published rows older than the retention age first
• an optional count cap then trims the oldest remaining published rows
• a practical starting point is 30 days of retention, with an optional count cap for high-volume systems

The SQLite table is expected to include:

• id
• message_type
• payload
• created_at
• claimed_at (nullable)
• published_at (nullable)

Example:

from dispatchbus.outbox import JSONSerializer, OutboxProcessor, OutboxRetentionPolicy, OutboxWorker

SQLite example:

from dispatchbus.outbox import SQLiteOutboxStorage

Ensuring atomicity

To guarantee that your outbox message is written if and only if your business data is saved, you should write both in the same transaction using SQLiteOutboxStorage.enqueue(). The enqueue() method inserts the outbox row using the configured connection but explicitly does not commit. You must manage the transaction and the final commit:

import aiosqlite

async def handle_request(payload: dict) -> None:
    async with aiosqlite.connect("database.db") as db:
        storage = SQLiteOutboxStorage(db)
        
        await db.execute("BEGIN")
        try:
            # 1. Write business data
            await db.execute("INSERT INTO users (name) VALUES (?)", (payload["name"],))
            
            # 2. Enqueue the outbox message
            event = UserCreated(name=payload["name"])
            await storage.enqueue(event, serializer)
            
            # 3. Commit both atomically
            await db.commit()
        except Exception:
            await db.rollback()
            raise

SQLite operational note: deleting rows does not necessarily shrink the database file immediately. If reclaiming file size matters, use SQLite operational tools such as VACUUM or configure auto-vacuum appropriately.

If aiosqlite is not installed, importing dispatchbus.outbox still works, but accessing SQLiteOutboxStorage raises an ImportError telling you to install dispatchbus[sqlite].

Public API

The package root currently exports:

• MessageBus
• MessageBase
• CommandBase
• EventBase
• MessageMetadata
• new_root_metadata
• derive_child_metadata
• get_metadata
• DispatchStarted
• DispatchFinished
• HandlerStarted
• HandlerFinished
• HandlerFailed
• DispatchbusError
• HandlerRegistrationError
• DuplicateCommandHandlerError
• NoCommandHandlerError
• EventPublicationError
• BusUsageError

Current limitations

dispatchbus is intentionally small today. Current limitations include:

• in-memory only; no broker, queue, or transport integration
• no built-in broker or transport integration
• no retries or scheduling support
• exact-type handler lookup only; no inheritance-based dispatch
• one command handler per command type
• sync handlers and sync subscribers run in a thread pool
• long-running blocking sync work can reduce throughput
• event ordering guarantees depend on the selected concurrency mode
• the dispatchbus CLI entry point is currently just a placeholder

Development

This repo uses uv, ruff, pyright, and pytest.

uv sync --dev
just format
just check

Available local commands:

just format
just lint
just typecheck
just test
just build
just check
just all