Skip to content

Basic usage

1. Declare the outbox table

The package never creates or migrates your schema; Alembic does that. 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. It also carries 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

Code that uses SQLAlchemy Core without the ORM can pass its AsyncConnection instead; see Publishing on a Core connection. Passing anything else 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, use 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