omni-box

Unified transactional messaging primitives for the Transactional Outbox and Transactional Inbox patterns in async Python services.

The library ships:

• Domain entities (OutboxEvent, InboxEvent) and a OmniBoxDomainService factory.
• Storage-agnostic repository protocols and a production-ready PostgreSQL implementation (SQLAlchemy 2 + asyncpg).
• A composable pipeline (EventProcessorBuilder, EventBatchProcessor) with built-in steps for metrics, OpenTelemetry, DLQ, circuit breaker and inbox deduplication.
• High-level orchestrators: OutboxPublisher (background publisher) and InboxConsumerRunner (broker consumer with configurable commit semantics).
• A pure-aiokafka broker adapter (publisher and consumer).

The library does not provide a Unit-of-Work implementation. The transactional boundary that ties business state with the outbox row (or with the inbox row plus side effects) is owned by the calling service.

Requirements

• Python 3.12+ (uses PEP 695 generics).
• The core package only depends on pydantic, orjson, and structlog.

Installation

pip install omni-box

Optional extras (declared under [project.optional-dependencies] in pyproject.toml):

Combine as needed, e.g. pip install "omni-box[postgres,kafka,metrics]".

Outbox Quick Start

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

# 1. Persist the event in the same DB transaction as your business state.
domain = OmniBoxDomainService()
event = domain.create_outbox_event(
    aggregate_type="user",
    aggregate_id=user_id,
    event_type="user.created",
    topic="users.events",
    partition_key=str(user_id),
    payload={"email": "user@example.com"},
)

async with uow.transaction() as tx:        # your own UoW, not part of omni-box
    await tx.users.create(user)
    await tx.outbox.create(event)

# 2. A background worker reads pending rows and publishes them, in a transaction of
#    its own. Nothing in omni-box commits: the fetch, the lock, the publish and the
#    status update are one unit of work, and it is yours to open and commit.
broker = KafkaEventPublisher(producer=kafka_producer, 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)

Forget that transaction and nothing happened: the lock and the completion roll back with the session, the rows stay pending, and the next cycle publishes them again.

OutboxPublisher is defined in omni_box.application.services.publish. Under the hood it builds an EventBatchProcessor via create_outbox_processor, so you get fetch, lock, retry, metrics and (optionally) DLQ for free.

Inbox Quick Start

InboxConsumerRunner consumes from a broker and lands every message in the inbox table inside a transaction. The transaction is opened via a user-supplied InboxTransactionProviderProtocol, which yields an InboxEventRepository bound to the open session — this keeps the library free of any UoW. The handler runs inside that transaction, and repo.session is how it writes its own side effects there.

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):
    """Bridges your session/UoW to the runner."""

    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_inbox_consumer,            # your EventConsumer adapter
    transaction_provider=InboxTxProvider(...),
    handler=handle_inbox_event,               # optional; runs inside the same tx
    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()

Commit semantics are configurable via ack_strategy (AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE_INBOX) and commit_offset_policy (ON_PERSIST, ON_SUCCESS). See omni_box.application.services.consume for the full contract.

Public API

The package re-exports its stable surface from omni_box:

import omni_box
print(omni_box.__all__)
print(omni_box.__version__)

For detailed component reference see docs/api_reference.md.

Documentation

• Architecture
• User guide
• Migrations & DDL
• Custom storage adapters
• Troubleshooting
• API reference
• For AI agents — the whole API surface, the rules that break code when broken and a map of the rest, on one page to hand to a coding assistant

License

Apache 2.0 — see LICENSE.