User Guide¶

omni-box provides production-ready primitives for Transactional Outbox and Transactional Inbox patterns in async Python services.

Installation¶

uv add omni-box
# or
pip install omni-box

Optional extras (see the extras table for the full list): postgres, kafka, metrics, opentelemetry, settings, dishka.

Basic concepts¶

Outbox — guarantees that an event reaches the broker iff the originating DB transaction committed. The publisher is a separate background job that reads pending rows.
Inbox — guarantees that each incoming broker message is handled at most once per (message_id, consumer_group) thanks to the inbox unique index, regardless of broker redelivery.

Transactional Outbox¶

1. Define your payload schema (optional)¶

from pydantic import BaseModel

class UserCreated(BaseModel):
    user_id: str
    email: str

2. Persist events inside your business transaction¶

from omni_box import OmniBoxDomainService

domain = OmniBoxDomainService()

async def create_user(uow, email: str):
    async with uow.transaction() as tx:
        user = await tx.users.create(email=email)
        event = domain.create_outbox_event(
            event_type="user.created",
            topic="users",
            partition_key=str(user.id),
            payload={"user_id": str(user.id), "email": email},
            aggregate_type="user",
            aggregate_id=user.id,
        )
        await tx.outbox.create(event)

tx.outbox is your service-owned repository — typically PostgresOutboxRepository bound to the same AsyncSession as the rest of your UoW.

3. Run the publisher¶

One cycle of publish_batch fetches and locks pending rows, publishes each of them, and writes the outcome back. All three happen through the session you handed the repository, and the library never commits: open a transaction per cycle and commit it yourself.

import asyncio

from omni_box import OutboxPublisher
from omni_box.core.converters import EnvelopeEventConverter
from omni_box.infra.brokers.kafka import KafkaEventPublisher
from omni_box.infra.storage.postgres import PostgresOutboxRepository

broker = KafkaEventPublisher(
    producer=kafka_producer,            # caller-owned AIOKafkaProducer
    converter=EnvelopeEventConverter(),
)

while not shutdown:
    async with session_factory() as session, session.begin():   # the commit is yours
        repo = PostgresOutboxRepository(session, model_class=OutboxEventDB)
        result = await OutboxPublisher(repo, broker).publish_batch(
            worker_id="publisher-1",
            batch_size=100,
        )
    if not result.processed_event_ids:
        await asyncio.sleep(1.0)

Drop the transaction and the cycle publishes for nothing: the lock and the completion roll back with the session, the rows are still pending, and the next cycle sends them again.

When the broker is down. A publish failure that is about the broker rather than about the row — the connection, the node, a request that timed out, the publish timeout itself — does not spend the row's attempt budget. KafkaEventPublisher raises TransientError once its own max_infra_retries are spent, and the outbox step records the row with attempts_made untouched and stops publishing for the rest of the cycle: the remaining rows are recorded the same way without being sent, since they were going to the same broker. So an outage costs one probe per cycle, the rows stay pending however long it lasts, and the first cycle after the broker answers publishes the backlog. A payload the broker rejects, or a topic it says it does not have, is about the row and still spends an attempt — that is what max_attempts and failed are for. Raise TransientError from your own publisher or handler to get the same treatment.

Transactional Inbox¶

Option A — drive consumption with `InboxConsumerRunner`¶

This is the typical "one Kafka message per transaction" loop with configurable commit semantics.

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from omni_box import AckStrategy, InboxConsumerRunner, InboxEvent
from omni_box.core.protocols import InboxEventRepository
from omni_box.core.protocols.transaction import InboxTransactionProviderProtocol


class InboxTxProvider(InboxTransactionProviderProtocol):
    def __init__(self, session_factory, repo_factory) -> None:
        self._session_factory = session_factory
        self._repo_factory = repo_factory

    @asynccontextmanager
    async def transaction(self) -> AsyncIterator[InboxEventRepository]:
        async with self._session_factory() as session, session.begin():
            yield self._repo_factory(session)


async def handle_inbox_event(event: InboxEvent, repo: InboxEventRepository) -> None:
    await repo.session.execute(             # the transaction the inbox row is in
        invoices.insert().values(order_id=event.payload["order_id"])
    )


runner = InboxConsumerRunner(
    consumer=kafka_consumer_adapter,
    transaction_provider=InboxTxProvider(session_factory, lambda s: PostgresInboxRepository(s, model_class=InboxEventDB)),
    handler=handle_inbox_event,     # optional; runs inside the same transaction
    worker_id="worker-1",
    consumer_group="identity-service",
    ack_strategy=AckStrategy.EXACTLY_ONCE_INBOX,
)

await runner.start()
try:
    await runner.run_forever()
finally:
    await runner.stop()

The handler runs inside the transaction that inserts the inbox row, and repo.session is that transaction. Write the side effect through it and the invoice and the inbox row commit together — or roll back together when the handler raises. A session the handler opens itself is a second transaction and does not get that.

Option B — batch processing already-stored inbox rows¶

Use create_inbox_processor when you want to ingest messages quickly (commit on persist) and process them in a separate workload.

from omni_box import InboxEvent, create_inbox_processor
from omni_box.core.protocols import InboxEventRepository
from omni_box.infra.storage.postgres import PostgresInboxRepository

async def my_handler(event: InboxEvent, repo: InboxEventRepository):
    print(f"Processing {event.event_type} ({event.message_id})")

async with session_factory() as session, session.begin():    # the commit is yours here too
    processor = create_inbox_processor(
        repo=PostgresInboxRepository(session, model_class=InboxEventDB),
        handler=my_handler,
        job_name="my_inbox_job",
    )
    await processor.process_batch(worker_id="worker-1", batch_size=50)

The batch processor owns the retry budget: a handler that fails leaves the row pending with attempts_made + 1 until max_attempts, which is why draining rows here — rather than in the runner's handler — is how you get retries out of the table.

Option C — automated event routing¶

EventRouter keys handlers by (topic, event_type, schema_version), and create_dispatching_processor dispatches on the event's source as the topic.

The @event_handler decorator only marks a method — it registers nothing on its own. Give the router a plain function through register_handler, or a BaseEventHandler subclass it can sweep:

from omni_box import (
    BaseEventHandler,
    EventRouter,
    InboxEvent,
    create_dispatching_processor,
    event_handler,
)
from omni_box.core.protocols import InboxEventRepository

router = EventRouter()


# … either a plain function, registered explicitly …
async def handle_user_deleted(event: InboxEvent, repo: InboxEventRepository, uow) -> None:
    ...


router.register_handler(event_type="user.deleted", topic="users", handler=handle_user_deleted)


# … or a class whose decorated methods the router sweeps.
class UserHandlers(BaseEventHandler):
    topic = "users"                      # the source the events arrive with

    @event_handler("user.created")       # optional: topic=..., schema_version=...
    async def on_created(self, event: InboxEvent, repo: InboxEventRepository, uow) -> None:
        ...


router.register_instance(UserHandlers())

processor = create_dispatching_processor(
    repo=inbox_repo,
    router=router,
    dependencies={"uow": uow},           # passed as keyword arguments to the handler
)

Extra keyword arguments named in dependencies are passed to every handler, so each handler signature must accept them. An event with no matching handler comes back as a failed EventHandlerResult reading No handler for topic=… event_type=… v=….

Building the pipeline yourself? create_dispatching_handler(router, **dependencies) from omni_box.core.dispatch is the router-backed handler this factory installs.

Customising the pipeline¶

Need full control? Build the processor yourself.

from omni_box import EventProcessorBuilder
from omni_box.core.pipeline.steps import (
    CircuitBreakerStep,
    DLQStep,
    HandlerExecutionStep,
    OpenTelemetryStep,
    PublisherExecutionStep,
    SiblingDeduplicationStep,
)
from omni_box.core.pipeline.strategies import (
    BulkCommitStrategy,
    DistributedLockingFetchStrategy,
)

builder = (
    EventProcessorBuilder(inbox_repo)
    .add_step(OpenTelemetryStep(service_name="my-service"))
    .add_step(CircuitBreakerStep(failure_threshold=5, recovery_timeout_seconds=60))
    .add_step(DLQStep(dlq_storage))
    .add_step(SiblingDeduplicationStep())
    .add_step(HandlerExecutionStep(my_handler, timeout=30))
    .with_fetch_strategy(DistributedLockingFetchStrategy())
    .with_commit_strategy(BulkCommitStrategy())
)

processor = builder.build()

The builder auto-picks DistributedLockingFetchStrategy + BulkCommitStrategy when the repository advertises matching capabilities, so the with_* calls above are usually optional.

For an outbox, use PublisherExecutionStep(broker.publish, timeout=30) in place of HandlerExecutionStep — it is what create_outbox_processor installs, and it is the step that keeps a broker outage off the attempt budget.

Outbox payload envelopes¶

OutboxPublisher delegates the body shape to a converter:

Converter	Body
`RawEventConverter`	Raw `event.payload`.
`SchemaVersionedConverter`	`{"schema_version": …, "payload": …}`.
`EnvelopeEventConverter`	Full envelope: payload + tracing identifiers.

from omni_box import EnvelopeEventConverter
from omni_box.infra.brokers.kafka import KafkaEventPublisher

broker = KafkaEventPublisher(producer=p, converter=EnvelopeEventConverter())

Observability¶

Structured logging¶

omni-box emits structured events through structlog. Configure renderers in your application bootstrap — for example:

import structlog

structlog.configure(
    processors=[
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]
)

omni-box does not ship a PII sanitiser; redact sensitive fields in your own structlog processor if needed.

Metrics¶

Pass metrics= to any factory or to MetricsStep. Either implement InboxMetrics / OutboxMetrics yourself or use the Prometheus adapters from omni_box.infra.metrics (extra: metrics).

OpenTelemetry¶

Add OpenTelemetryStep(service_name="my-service") to your pipeline (extra: opentelemetry).

Dead Letter Queue¶

from omni_box.core.pipeline.steps import DLQStep, DLQStorage

class MyDLQStorage(DLQStorage):
    async def move_to_dlq(self, event, error: str) -> None:
        await db.save_to_dlq(event, error)

builder.add_step(DLQStep(MyDLQStorage()))

DLQStep is best-effort and non-transactional. Transient failures (count_as_attempt=False) are never routed to DLQ.

Circuit breaker¶

from omni_box.core.pipeline.steps import CircuitBreakerStep

builder.add_step(CircuitBreakerStep(failure_threshold=5, recovery_timeout_seconds=60))

The breaker state is held in-process — it does not survive restarts and is not shared between replicas. For multi-worker deployments add an external coordination layer (e.g. Redis-backed counters) on top of this step.

Maintenance¶

Run periodically against the same repository:

from omni_box import OmniBoxMaintenanceService

m = OmniBoxMaintenanceService(repo=outbox_repo)
await m.release_stale_locks(stale_timeout_seconds=300)
await m.cleanup_old_events(retention_days=14)

Both methods require the repository to implement SupportsRetentionPolicies. PostgresOutboxRepository and PostgresInboxRepository do.