natsbox

Transactional Outbox Pattern for Python — PostgreSQL + NATS JetStream

📚 Full Documentation is available at https://ademboukabes.github.io/natsbox/

What problem does this solve?

In event-driven architectures, a service typically needs to do two things atomically:

1. Persist a domain state change to PostgreSQL (e.g. insert an Order row).
2. Publish a corresponding event to NATS JetStream (e.g. order.created).

Doing this without a distributed transaction exposes your system to silent inconsistencies (events lost if the app crashes before publishing, or phantom events sent if the DB rolls back).

nats-outbox solves this with the Transactional Outbox Pattern. It writes events to a dedicated outbox_events table in the same SQL transaction as your domain data. A separate relay process reads this table and publishes events reliably to NATS JetStream, guaranteeing at-least-once delivery.

Quick Start

pip install nats-outbox[cli,all]

from sqlalchemy.ext.asyncio import AsyncSession
from nats_outbox.core.outbox import outbox_transaction

async def create_order(session: AsyncSession, user_id: int, amount: float):
    # This block automatically commits your session on success!
    async with outbox_transaction(session) as tx:
        order = Order(user_id=user_id, amount=amount)
        tx.add(order)
        await session.flush()  # get order.id before commit

        # Stage the event. It is saved in PostgreSQL atomically with your Order.
        tx.publish_event(
            subject="order.created",
            payload={"order_id": str(order.id), "amount": amount},
            aggregate_id=str(order.id),
            aggregate_type="Order",
        )

Start the background relay process to push events to NATS:

OUTBOX_DATABASE_URL=postgresql+asyncpg://... \
OUTBOX_NATS_URL=nats://localhost:4222 \
nats-outbox relay start

Documentation

Please visit the official documentation for the complete Integration Guide, Architecture details, Schema definitions, and Relay Configuration Reference.

License

MIT