Skip to content

Basic usage

1. Declare the outbox table

The package never creates or migrates your schema — that's Alembic's job. make_outbox_table(metadata, table_name="outbox") returns a sqlalchemy.Table you attach to your own MetaData:

from sqlalchemy import MetaData
from faststream_outbox import make_outbox_table

metadata = MetaData()
outbox_table = make_outbox_table(metadata, table_name="outbox")

The returned Table carries three indexes the broker needs at runtime — a partial index for the fetch CTE's unleased branch, a partial index for the expired-lease reclaim branch, and a partial unique index for timer_id deduplication — plus a CHECK ((acquired_token IS NULL) = (acquired_at IS NULL)) constraint that makes a half-set lease unrepresentable. Alembic autogenerate picks them all up alongside the table itself.

2. Create the broker and app

from sqlalchemy.ext.asyncio import create_async_engine
from faststream import FastStream
from faststream_outbox import OutboxBroker

engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost:5432/outbox")
broker = OutboxBroker(engine, outbox_table=outbox_table)
app = FastStream(broker)

3. Register a subscriber

Subscribers work like any FastStream subscriber. Decorate a handler with @broker.subscriber(queue, ...):

@broker.subscriber("orders")
async def handle(order_id: int) -> None:
    print(f"order {order_id}")

See Subscriber for the full options list (concurrency via max_workers, tuning, retry strategies).

4. Publish a message

broker.publish(body, *, queue, session, ...) inserts an outbox row through the caller's AsyncSession. It does not flush, commit, or open its own transaction — the row commits with the caller's domain writes:

from sqlalchemy.ext.asyncio import async_sessionmaker

session_factory = async_sessionmaker(engine, expire_on_commit=False)

async with session_factory() as session, session.begin():
    session.add(Order(id=1))  # `Order` is your own ORM model — the domain write
    await broker.publish(1, queue="orders", session=session)
    # session.begin() commits both atomically on exit

Passing anything that is not an AsyncSession raises TypeError. The whole point of the outbox pattern is that the row commits atomically with your domain writes; opening a separate session would defeat it.

See Publisher for publish_batch, the OutboxPublisher decorator, and chained publishing via OutboxResponse.

Full quickstart

from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from faststream import FastStream
from faststream_outbox import OutboxBroker, make_outbox_table

metadata = MetaData()
outbox_table = make_outbox_table(metadata, table_name="outbox")

engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost:5432/outbox")
broker = OutboxBroker(engine, outbox_table=outbox_table)
app = FastStream(broker)


@broker.subscriber("orders")
async def handle(order_id: int) -> None:
    print(f"order {order_id}")


session_factory = async_sessionmaker(engine, expire_on_commit=False)


@app.after_startup
async def publish_one() -> None:
    async with session_factory() as session, session.begin():
        await broker.publish(1, queue="orders", session=session)

To actually run this you need the asyncpg driver and the FastStream CLI (the base install ships neither):

pip install 'faststream-outbox[asyncpg]' 'faststream[cli]'

See Installation for the full extras list and a one-line Postgres container.

The package declares the outbox table but never creates it. Create it once before the first run — for local dev, a one-shot metadata.create_all against the same metadata/engine from above:

async with engine.begin() as conn:
    await conn.run_sync(metadata.create_all)

Production uses Alembic instead (see Alembic migrations); the first-outbox-app tutorial shows the full create_schema.py script. Once the table exists, save the module as app.py and run it with faststream run app:app.

Connection ownership

OutboxBroker does not close the AsyncEngine you pass in — the caller owns its lifecycle. The same engine can be shared with other SQLAlchemy users (your FastAPI app, an Alembic upgrade, etc.); closing it from the broker would surprise them. Manage the engine with try/finally or — when running under FastAPI — let the framework's lifespan handle it (see FastAPI integration).

  • How it works — architecture, lease invariant, at-least-once semantics
  • Subscriber — tuning, retry strategies, slow-handler queue segregation
  • Publisher — publish_batch, OutboxPublisher, chained publishing
  • FastAPI integration — OutboxRouter, Depends(get_session) pattern