Metadata-Version: 2.4
Name: dispatchbus
Version: 0.1.0a1
Summary: Dispatchbus Python project.
Project-URL: Homepage, https://github.com/kevin-rieck/dispatchbus
Project-URL: Repository, https://github.com/kevin-rieck/dispatchbus.git
Project-URL: Issues, https://github.com/kevin-rieck/dispatchbus/issues
Author: Kevin Rieck
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.12
Provides-Extra: sqlite
Requires-Dist: aiosqlite<1,>=0.20; extra == 'sqlite'
Description-Content-Type: text/markdown

# 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`:

```powershell
uv add dispatchbus
```

With `pip`:

```powershell
pip install dispatchbus
```

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

```powershell
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

```python
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:

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


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

The context lets a handler emit follow-up events:

```python
context.emit(SomeEvent(...))
```

## Event concurrency

Configure event execution with `event_concurrency`:

```python
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.

```python
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`

```python
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(...)`

```python
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:

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

SQLite example:

```python
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:

```python
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`.

```powershell
uv sync --dev
just format
just check
```

Available local commands:

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