FastStream Outbox

https://github.com/modern-python/faststream-outbox
Admin
FastStream broker integration for the transactional outbox pattern, using a Postgres table as the...
Tokens:65,996
Snippets:463
Trust Score:9
License:MIT
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install dependencies for quickstart

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/basic.md

Install the required asyncpg driver and FastStream CLI to run the quickstart example. The base install does not include these dependencies.

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

--------------------------------

### Full quickstart example

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/basic.md

Complete example showing how to set up the outbox broker, define a subscriber, and publish a message after startup. Requires the asyncpg driver and FastStream CLI, which are not included in the base install.

```python
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)
```

--------------------------------

### FastAPI outbox quickstart example

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/fastapi.md

Complete example showing how to set up an OutboxRouter with FastAPI, sharing an AsyncSession between HTTP routes and outbox subscribers. The router auto-starts the broker via FastAPI's lifespan; do not call broker.start().

```python
from collections.abc import AsyncIterator

from fastapi import Depends, FastAPI
from pydantic import BaseModel
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from faststream_outbox import make_outbox_table
from faststream_outbox.fastapi import OutboxRouter


class OrderIn(BaseModel):
    item: str


class Base(DeclarativeBase):
    pass


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    item: Mapped[str]


metadata = MetaData()
outbox_table = make_outbox_table(metadata, table_name="outbox")
engine = create_async_engine("postgresql+asyncpg://localhost/app")
session_factory = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with session_factory() as s, s.begin():
        yield s


router = OutboxRouter(engine, outbox_table=outbox_table)


@router.subscriber("orders")
async def handle(
    body: dict,
    session: AsyncSession = Depends(get_session),
) -> None: ...  # domain writes on `session` commit with any chained outbox publishes


@router.post("/orders")
async def create_order(
    order: OrderIn,
    session: AsyncSession = Depends(get_session),
) -> dict:
    db_order = Order(item=order.item)
    session.add(db_order)
    await session.flush()  # populate db_order.id
    await router.broker.publish({"order_id": db_order.id}, queue="orders", session=session)
    return {"ok": True}


app = FastAPI()
app.include_router(router)
```

--------------------------------

### Standalone: FastStream app with KafkaBroker connect on_startup

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/relay.md

Alternative setup using a single FastStream app. The foreign broker's connect is hooked into on_startup. The outbox broker logs a WARNING per unstarted foreign broker at start(), and a relay to an unstarted broker fails-and-retries until it starts.

```python
from faststream import FastStream
from faststream.kafka import KafkaBroker
from faststream_outbox import OutboxBroker

broker_outbox = OutboxBroker(engine, outbox_table=outbox_table)
broker_kafka = KafkaBroker("127.0.0.1:9092")
publisher_kafka = broker_kafka.publisher("kafka_topic")


@publisher_kafka
@broker_outbox.subscriber("outbox_queue")
async def relay(body: dict) -> dict:
    return body


app = FastStream(broker_outbox, on_startup=[broker_kafka.connect])
```

--------------------------------

### Install and configure Prometheus adapter with OutboxBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/setup-prometheus-opentelemetry.md

Install the Prometheus adapter and run the example app with uvicorn. The Python snippet shows how to configure OutboxBroker with PrometheusRecorder and expose /metrics and /healthz endpoints via AsgiFastStream.

```bash
pip install 'faststream-outbox[prometheus]' uvicorn
```

```python
# app.py — run with `uvicorn app:app --host 0.0.0.0 --port 8000`
from faststream.asgi import AsgiFastStream, make_ping_asgi
from prometheus_client import REGISTRY, make_asgi_app
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine

from faststream_outbox import OutboxBroker, make_outbox_table
from faststream_outbox.metrics.prometheus import PrometheusRecorder


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,
    metrics_recorder=PrometheusRecorder(app_name="checkout", registry=REGISTRY),
)


@broker.subscriber("orders", max_workers=4)
async def handle_order(body: dict) -> None: ...


app = AsgiFastStream(
    broker,
    asgi_routes=[
        ("/metrics", make_asgi_app(registry=REGISTRY)),
        ("/healthz", make_ping_asgi(broker, timeout=2.0)),
    ],
)
```

--------------------------------

### Install all extras

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Installs every optional extra at once: asyncpg, validate, fastapi, prometheus, opentelemetry.

```bash
pip install 'faststream-outbox[all]'
```

--------------------------------

### Output of uv add faststream[cli,kafka]

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Example output from the uv add command, showing the resolved and installed packages. The exact versions shown are illustrative; your pinned versions will differ.

```text
Resolved 29 packages in 785ms
Installed 3 packages in 6ms
 + aiokafka==0.14.0
 + async-timeout==5.0.1
 + packaging==26.2
```

--------------------------------

### Install prometheus extra

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Installs the PrometheusRecorder metrics adapter and native OutboxPrometheusMiddleware.

```bash
pip install 'faststream-outbox[prometheus]'
```

--------------------------------

### Install fastapi extra

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Installs the FastAPI integration, enabling the OutboxRouter.

```bash
pip install 'faststream-outbox[fastapi]'
```

--------------------------------

### FastAPI: Mount OutboxRouter and KafkaRouter on same app

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/relay.md

Recommended setup. Both routers auto-start via FastAPI's lifespan. The outbox broker logs a WARNING per unstarted foreign broker at start(), and a relay to an unstarted broker fails-and-retries until it starts.

```python
from fastapi import FastAPI
from faststream.kafka.fastapi import KafkaRouter
from faststream_outbox.fastapi import OutboxRouter

outbox_router = OutboxRouter(engine, outbox_table=outbox_table)
kafka_router = KafkaRouter("127.0.0.1:9092")

publisher_kafka = kafka_router.publisher("kafka_topic")


@publisher_kafka
@outbox_router.subscriber("outbox_queue")
async def relay(body: dict) -> dict:
    return body


app = FastAPI()
app.include_router(outbox_router)
app.include_router(kafka_router)
```

--------------------------------

### Install opentelemetry extra

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Installs the OpenTelemetryRecorder metrics adapter and native OutboxTelemetryMiddleware.

```bash
pip install 'faststream-outbox[opentelemetry]'
```

--------------------------------

### Install and configure OpenTelemetry + Prometheus with OutboxBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/setup-prometheus-opentelemetry.md

Install the required packages for OpenTelemetry and Prometheus integration, then run the application with the OTLP endpoint and uvicorn. The Python example sets up an OutboxBroker with both the OTel middleware (span-only) and Prometheus middleware, plus a PrometheusRecorder for outbox-internal events, using separate registries to avoid duplicate collector errors.

```bash
pip install 'faststream-outbox[opentelemetry,prometheus]' \
    opentelemetry-exporter-otlp uvicorn
```

```python
# app.py — run with `OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \
#                    uvicorn app:app --host 0.0.0.0 --port 8000`
from faststream.asgi import AsgiFastStream
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from prometheus_client import CollectorRegistry, make_asgi_app
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine

from faststream_outbox import OutboxBroker, make_outbox_table
from faststream_outbox.metrics.prometheus import PrometheusRecorder
from faststream_outbox.opentelemetry import OutboxTelemetryMiddleware
from faststream_outbox.prometheus import OutboxPrometheusMiddleware


# ----- OTel SDK -----
resource = Resource.create({"service.name": "my-outbox-service"})
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(tracer_provider)

# Two registries: the middleware and the recorder both define the same
# faststream_* consume/publish collectors, so sharing one registry raises
# "Duplicated timeseries in CollectorRegistry" at broker construction.
MIDDLEWARE_REGISTRY = CollectorRegistry()
RECORDER_REGISTRY = CollectorRegistry()

# ----- Outbox broker -----
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,
    middlewares=[
        # Bus-scope spans around consume_scope / publish_scope (span-only; see note above).
        OutboxTelemetryMiddleware(tracer_provider=tracer_provider),
        OutboxPrometheusMiddleware(registry=MIDDLEWARE_REGISTRY, app_name="my-outbox-service"),
    ],
    # Outbox-internal events (fetched, lease_lost, terminal reasons, dlq_written)
    # that have no message context and can't reach the middleware bus.
    metrics_recorder=PrometheusRecorder(registry=RECORDER_REGISTRY, app_name="my-outbox-service"),
)


@broker.subscriber("orders", max_workers=4)
async def handle_order(body: dict) -> None: ...


app = AsgiFastStream(
    broker,
    asgi_routes=[
        ("/metrics", make_asgi_app(registry=MIDDLEWARE_REGISTRY)),
        ("/metrics/outbox", make_asgi_app(registry=RECORDER_REGISTRY)),
    ],
)
```

--------------------------------

### Start Postgres container and verify readiness

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Starts a disposable Postgres 17 container with credentials outbox/outbox/outbox, then verifies readiness via logs. The container ID and the final log line are shown as expected output.

```bash
docker run --rm -d --name outbox-postgres \
    -e POSTGRES_USER=outbox -e POSTGRES_PASSWORD=outbox -e POSTGRES_DB=outbox \
    -p 5432:5432 postgres:17
```

```text
7558ba0b8949e6410415f51152cd2da9b5eaab4ebae092aa14f2a6094f57d98f
```

```bash
docker logs outbox-postgres 2>&1 | tail -1
```

```text
2026-06-12 05:05:44.529 UTC [1] LOG:  database system is ready to accept connections
```

--------------------------------

### Install asyncpg extra

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Installs the asyncpg SQLAlchemy driver, required for LISTEN/NOTIFY short-circuit wakeups in the subscriber's fetch loop.

```bash
pip install 'faststream-outbox[asyncpg]'
```

--------------------------------

### Full app.py with outbox setup and handlers

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Complete app.py after adding the subscriber and the after-startup publisher. It sets up the outbox table, engine, broker, and session factory, then defines the handler and the publish hook.

```python
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)

session_factory = async_sessionmaker(engine, expire_on_commit=False)


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


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

--------------------------------

### Install faststream-outbox with pip

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Installs faststream-outbox using pip.

```bash
pip install faststream-outbox
```

--------------------------------

### Install faststream[cli,kafka] with uv

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Installs the FastStream CLI and Kafka dependencies using uv. The output shows the resolved and installed packages, including aiokafka and async-timeout, though your pinned versions will differ.

```bash
uv add 'faststream[cli,kafka]'
```

--------------------------------

### Check Kafka broker startup log

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Greps the Kafka container logs for the 'Kafka Server started' message to confirm the broker came up cleanly.

```bash
docker compose logs kafka 2>&1 | grep -m1 'Kafka Server started'
```

--------------------------------

### Install multiple extras with commas

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Combines multiple optional extras in a single pip install command.

```bash
pip install 'faststream-outbox[asyncpg,fastapi,prometheus]'
```

--------------------------------

### Install faststream-outbox with uv

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Adds faststream-outbox to the project using the uv package manager.

```bash
uv add faststream-outbox
```

--------------------------------

### Start Postgres with Docker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Runs a Postgres 17 container with user, password, and database set to 'outbox', exposing port 5432.

```bash
docker run -d -p 5432:5432 \
    -e POSTGRES_USER=outbox \
    -e POSTGRES_PASSWORD=outbox \
    -e POSTGRES_DB=outbox \
    postgres:17
```

--------------------------------

### Expected Kafka startup log line

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Shows the expected log line indicating the Kafka broker has started successfully.

```text
outbox-kafka  | [2026-06-12 05:22:33,782] INFO [KafkaRaftServer nodeId=1] Kafka Server started (kafka.server.KafkaRaftServer)
```

--------------------------------

### Startup log output showing handler firing

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Expected output after starting the app. The line 'got order 1' indicates the handler fired after the outbox row was inserted by the @app.after_startup hook.

```text
2026-06-12 08:07:06,179 INFO     - FastStream app starting...
2026-06-12 08:07:06,179 INFO     - orders  |  - `Handle` waiting for messages
2026-06-12 08:07:06,276 INFO     - FastStream app started successfully! To exit, press CTRL+C
2026-06-12 08:07:06,283 INFO     - orders  |  - Received
got order 1
2026-06-12 08:07:06,283 INFO     - orders  |  - Processed
```

--------------------------------

### Start Kafka container

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Starts the Kafka container defined in docker-compose.yml. The expected output shows the network and container creation steps.

```bash
docker compose up -d kafka
```

--------------------------------

### Stop the outbox PostgreSQL container

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Stops the PostgreSQL container started for the outbox tutorial. Run this when you are finished working with the example app.

```bash
docker stop outbox-postgres
```

--------------------------------

### Install faststream-outbox with poetry

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Adds faststream-outbox to the project using the Poetry package manager.

```bash
poetry add faststream-outbox
```

--------------------------------

### Standalone router include order

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/relay.md

For the non-FastAPI lifecycle, include both routers before starting the brokers. The comment notes that kafka_router and outbox_router are constructed as in the FastAPI example (KafkaRouter(...) / OutboxRouter(...)).

```python
# kafka_router / outbox_router are constructed exactly as in the FastAPI
# example above (KafkaRouter(...) / OutboxRouter(...)).
broker_kafka.include_router(kafka_router)
broker_outbox.include_router(outbox_router)
# then start
```

--------------------------------

### Install OpenTelemetry and Prometheus dependencies

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/setup-prometheus-opentelemetry.md

Installs the required packages for the OpenTelemetry adapter with Prometheus export, including the optional dependencies for OpenTelemetry and Prometheus, the Prometheus exporter, and uvicorn for running the ASGI app.

```bash
pip install 'faststream-outbox[opentelemetry,prometheus]' \
    opentelemetry-exporter-prometheus uvicorn
```

--------------------------------

### Run the FastStream app

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Starts the FastStream application in a terminal. The expected output shows the app starting, the orders handler waiting for messages, and a received message being processed.

```bash
uv run faststream run app:app
```

--------------------------------

### Full app.py with Kafka relay

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

The full app.py after adding the Kafka broker and publisher. It includes the outbox setup, the Kafka publisher, the stacked handler, and an after_startup hook that publishes a test order.

```python
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from faststream import FastStream
from faststream.kafka import KafkaBroker
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_outbox = OutboxBroker(engine, outbox_table=outbox_table)
broker_kafka = KafkaBroker("localhost:9092")
kafka_publisher = broker_kafka.publisher("orders.kafka")

app = FastStream(broker_outbox, on_startup=[broker_kafka.connect])

session_factory = async_sessionmaker(engine, expire_on_commit=False)


@kafka_publisher
@broker_outbox.subscriber("orders")
async def handle(order_id: int) -> int:
    print(f"got order {order_id}")
    return order_id


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

--------------------------------

### Install validate extra

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/installation.md

Installs Alembic for broker.validate_schema(). Calling validate_schema() without this extra raises ImportError.

```bash
pip install 'faststream-outbox[validate]'
```

--------------------------------

### Expected output from docker compose up

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Shows the expected output when starting the Kafka container, including network and container creation messages.

```text
 Network outbox-tutorial_default  Creating
 Network outbox-tutorial_default  Created
 Container outbox-kafka  Creating
 Container outbox-kafka  Created
 Container outbox-kafka  Starting
 Container outbox-kafka  Started
```

--------------------------------

### Configuring OutboxRouter with DLQ and metrics_recorder

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/fastapi.md

Shows how to pass “dlq_table” and “metrics_recorder” to “OutboxRouter(...)”, which forward to the inner “OutboxBroker”. “make_dlq_table” is imported alongside “make_outbox_table”. The example uses “engine” and “outbox_table” variables defined elsewhere on the page.

```python
from faststream_outbox import make_dlq_table  # alongside make_outbox_table

outbox_router = OutboxRouter(
    engine,
    outbox_table=outbox_table,
    dlq_table=make_dlq_table(metadata),
    metrics_recorder=my_recorder,
)

```

--------------------------------

### PrometheusRecorder setup for OutboxBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/observability.md

Configures OutboxBroker with the bundled PrometheusRecorder, passing an app name and a CollectorRegistry. This emits the faststream_outbox_* series used by the PromQL playbook. The recorder is called from the event loop, so it must not block.

```python
from prometheus_client import CollectorRegistry
from faststream_outbox.metrics.prometheus import PrometheusRecorder

registry = CollectorRegistry()
broker = OutboxBroker(
    engine,
    outbox_table=outbox_table,
    metrics_recorder=PrometheusRecorder(app_name="checkout", registry=registry),
)
```

--------------------------------

### Minimal Kafka relay with OutboxBroker and KafkaBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/relay.md

Sets up an OutboxBroker with an engine and outbox table, a KafkaBroker pointing at 127.0.0.1:9092, and a Kafka publisher for 'kafka_topic'. The relay subscriber listens on 'outbox_queue' and returns the body, which the publisher decorator then sends to Kafka. Assumes 'engine' and 'outbox_table' from the Basic usage setup.

```python
from faststream.kafka import KafkaBroker
from faststream_outbox import OutboxBroker

broker_outbox = OutboxBroker(engine, outbox_table=outbox_table)
broker_kafka = KafkaBroker("127.0.0.1:9092")
publisher_kafka = broker_kafka.publisher("kafka_topic")


@publisher_kafka
@broker_outbox.subscriber("outbox_queue")
async def relay(body: dict) -> dict:
    return body
```

--------------------------------

### Expected app startup output

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Expected console output after starting the app. The 'Topic orders.kafka not found in cluster metadata' line is aiokafka noticing a brand-new topic and asking the broker to auto-create it; it appears only on first run.

```text
2026-06-12 08:23:28,284 INFO     - FastStream app starting...
2026-06-12 08:23:28,328 INFO     - orders  |  - `Handle` waiting for messages
2026-06-12 08:23:28,389 INFO     - FastStream app started successfully! To exit, press CTRL+C
2026-06-12 08:23:28,394 INFO     - orders  |  - Received
Topic orders.kafka not found in cluster metadata
got order 1
2026-06-12 08:23:28,527 INFO     - orders  |  - Processed
```

--------------------------------

### TestOutboxBroker DLQ test example

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/dlq.md

Use TestOutboxBroker to simulate a DLQ failure and assert on the fake client's dlq_rows. The broker is bound to a name, and the fake mirrors the production CTE side-effect: the source row is removed from fake_client.rows and an audit dict is appended to fake_client.dlq_rows in the same call. The example publishes a message that raises a RuntimeError, then checks that the source row is gone and the DLQ row records a 'retry_terminal' failure.

```python
from sqlalchemy import MetaData

from faststream_outbox import NoRetry, OutboxBroker, TestOutboxBroker, make_dlq_table, make_outbox_table


metadata = MetaData()
outbox_table = make_outbox_table(metadata, table_name="outbox")
dlq_table = make_dlq_table(metadata, table_name="outbox_dlq")
broker = OutboxBroker(outbox_table=outbox_table, dlq_table=dlq_table)


@broker.subscriber("orders", retry_strategy=NoRetry())
async def handle(body: dict) -> None:
    raise RuntimeError("boom")


test_broker = TestOutboxBroker(broker)
async with test_broker:
    await broker.publish({"order_id": 1}, queue="orders")

assert test_broker.fake_client.rows == []
assert len(test_broker.fake_client.dlq_rows) == 1
assert test_broker.fake_client.dlq_rows[0]["failure_reason"] == "retry_terminal"
```

--------------------------------

### Create OutboxBroker and FastStream app

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/basic.md

Sets up the async engine, OutboxBroker, and FastStream app. The broker requires the engine and the outbox_table created earlier.

```python
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)
```

--------------------------------

### Configure DLQ table and broker with make_dlq_table

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/dlq.md

Builds the DLQ Table on the same MetaData as the outbox table and wires it into the OutboxBroker. The package does not create or migrate the table; run metadata.create_all or an Alembic migration after both tables are declared. Subscribers and publishers need no further configuration.

```python
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine

from faststream_outbox import OutboxBroker, make_dlq_table, make_outbox_table


metadata = MetaData()
outbox_table = make_outbox_table(metadata, table_name="outbox")
dlq_table = make_dlq_table(metadata, table_name="outbox_dlq")

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

--------------------------------

### Run create_schema.py with uv

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Runs the schema creation script. Expected to produce no output on success.

```bash
uv run python create_schema.py
```

--------------------------------

### Create outbox table with metadata.create_all

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/basic.md

Create the outbox table once before the first run. The package declares the table but never creates it. This one-shot command is for local development; production uses Alembic migrations.

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

--------------------------------

### Create outbox schema with create_schema.py

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

One-shot script that creates the outbox table by running metadata.create_all. Uses an async SQLAlchemy engine and the make_outbox_table helper. Self-contained for the tutorial; real projects would share metadata from a module.

```python
import asyncio

from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine

from faststream_outbox import make_outbox_table

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


async def main() -> None:
    engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost:5432/outbox")
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)
    await engine.dispose()


asyncio.run(main())
```

--------------------------------

### Define a subscriber for the 'orders' queue

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Adds a subscriber to the 'orders' queue. The handler runs once per row published to the queue, printing the order ID. No command is needed; it executes when a row is published and the app starts.

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

--------------------------------

### Declare outbox table and configure FastStream app

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Creates app.py to set up SQLAlchemy MetaData, declare the outbox table via make_outbox_table, build an async engine, and wire the broker and FastStream app. The package does not create or migrate the schema automatically.

```python
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)

session_factory = async_sessionmaker(engine, expire_on_commit=False)
```

--------------------------------

### Subscriber with outbox-specific options

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/subscriber.md

Configure per-subscriber knobs like max_workers, fetch_batch_size, and lease_ttl_seconds. Validation in OutboxSubscriberConfig.__post_init__ checks for likely-wrong combinations across all construction paths.

```python
@broker.subscriber(
    "high-priority",
    max_workers=8,
    fetch_batch_size=50,
    min_fetch_interval=0.1,
    max_fetch_interval=1.0,
    lease_ttl_seconds=120.0,
)
async def handle_urgent(body: dict) -> None: ...
```

--------------------------------

### Basic test with TestOutboxBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/testing.md

Shows a basic async test using TestOutboxBroker. Assumes asyncio_mode = "auto" so no pytest marker is needed. The broker is created with engine=None and a SQLAlchemy outbox table. Handlers run synchronously inside publish, so no wait is needed.

```python
from faststream_outbox import OutboxBroker, make_outbox_table
from faststream_outbox.testing import TestOutboxBroker
from sqlalchemy import MetaData


async def test_handler() -> None:
    metadata = MetaData()
    outbox_table = make_outbox_table(metadata, table_name="outbox")
    broker = OutboxBroker(None, outbox_table=outbox_table)  # engine not needed
    received: list[int] = []

    @broker.subscriber("orders")
    async def handle(order_id: int) -> None:
        received.append(order_id)

    async with TestOutboxBroker(broker):
        await broker.publish(1, queue="orders")
        # Handler has already run.

    assert received == [1]
```

--------------------------------

### Convert outbox_dlq to range-partitioned table (Alembic migration)

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/operations/alembic.md

One-time Alembic migration that converts the outbox_dlq table to a range-partitioned table by failed_at. It renames the existing table, creates the partitioned parent with a composite primary key (id, failed_at), adds an index on (queue, failed_at), creates initial partitions for May and June 2026, moves surviving rows, and drops the old table. validate_schema() continues to work because Alembic's autogenerate ignores partition boundaries when comparing column shape.

```python
# ### Convert outbox_dlq to range-partitioned by failed_at ###

# Rename the existing table out of the way.
op.rename_table("outbox_dlq", "outbox_dlq_old")

# Create the partitioned parent. The PRIMARY KEY must include the
# partition key, so we extend it to (id, failed_at).
op.execute("""
    CREATE TABLE outbox_dlq (
        id              BIGSERIAL NOT NULL,
        original_id     BIGINT NOT NULL,
        queue           VARCHAR(255) NOT NULL,
        payload         BYTEA NOT NULL,
        headers         JSONB,
        deliveries_count BIGINT NOT NULL,
        created_at      TIMESTAMPTZ NOT NULL,
        failed_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
        failure_reason  VARCHAR(64) NOT NULL,
        last_exception  TEXT,
        timer_id        VARCHAR(255),
        PRIMARY KEY (id, failed_at)
    ) PARTITION BY RANGE (failed_at);
""")

op.execute("""
    CREATE INDEX outbox_dlq_queue_failed_idx
        ON outbox_dlq (queue, failed_at);
""")

# Create initial partitions covering recent + current month.
op.execute("""
    CREATE TABLE outbox_dlq_2026_05 PARTITION OF outbox_dlq
        FOR VALUES FROM ('2026-05-01') TO ('2026-06-01');
    CREATE TABLE outbox_dlq_2026_06 PARTITION OF outbox_dlq
        FOR VALUES FROM ('2026-06-01') TO ('2026-07-01');
""")

# Move surviving rows into the partitioned table; rows older than the
# retention window stay in outbox_dlq_old (or get dropped explicitly).
op.execute("""
    INSERT INTO outbox_dlq
        SELECT * FROM outbox_dlq_old
        WHERE failed_at >= '2026-05-01';
""")

op.drop_table("outbox_dlq_old")
```

--------------------------------

### Publish a row with broker.publish in after_startup hook

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Adds an after-startup hook that publishes one row to the 'orders' queue. The publish inserts an outbox row through the provided session, which commits with the surrounding transaction. There is no separate send step; the commit is the send.

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

--------------------------------

### Inspect fake_client.rows with TestOutboxBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/testing.md

Shows how to bind the TestOutboxBroker harness to a name to inspect fake_client.rows. Uses a queue without a subscriber so the row survives; a queue with a subscriber would be dispatched and deleted, leaving rows empty.

```python
tb = TestOutboxBroker(broker)
async with tb:
    # "orders" has a subscriber, so in sync mode the row is dispatched and
    # acked (deleted) before publish returns. Publish to a queue with no
    # subscriber to inspect a row that survives.
    await broker.publish(1, queue="audit")

assert len(tb.fake_client.rows) == 1
```

--------------------------------

### Test composed app with TestOutboxBroker and TestKafkaBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/messaging-service.md

Nests TestOutboxBroker and TestKafkaBroker in sync mode so broker.publish drives the subscriber in-process; one use-case call runs the whole chain (outbox row → relay handler → Kafka) and you assert on the Kafka test broker without a background loop. Future-dated rows fire immediately in sync mode, so the 30-second unread-timers row is dispatched at once; to test the delay and cancel race for real, construct TestOutboxBroker(outbox_broker, run_loops=True).

```python
from faststream.kafka import TestKafkaBroker
from faststream_outbox import TestOutboxBroker


async def test_create_message_relays_event_to_kafka(
    outbox_broker,
    kafka_broker,
    create_message_use_case,
    command,
    kafka_publisher,
) -> None:
    async with TestOutboxBroker(outbox_broker), TestKafkaBroker(kafka_broker):
        await create_message_use_case(command)

        kafka_publisher.mock.assert_called_once()
```

--------------------------------

### Create an OutboxRouter and subscribe to a queue

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/router.md

Creates an OutboxRouter and registers a subscriber for the "orders" queue. The handler prints the received order_id.

```python
from faststream_outbox import OutboxRouter


router = OutboxRouter()


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

--------------------------------

### Configure OutboxBroker with DLQ table

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/introduction/how-it-works.md

Configure an OutboxBroker with a dead-letter queue table so terminal failures are copied into a sibling audit table in the same Postgres statement as the DELETE. Requires a SQLAlchemy async engine and the faststream_outbox helpers make_outbox_table and make_dlq_table.

```python
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine
from faststream_outbox import OutboxBroker, make_dlq_table, make_outbox_table

metadata = MetaData()
outbox_table = make_outbox_table(metadata, table_name="outbox")
dlq_table = make_dlq_table(metadata, table_name="outbox_dlq")
engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost:5432/outbox")
broker = OutboxBroker(engine, outbox_table=outbox_table, dlq_table=dlq_table)
```

--------------------------------

### Declare outbox table with make_outbox_table

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/basic.md

Creates a SQLAlchemy Table for the outbox pattern. The package does not migrate the schema; use Alembic. The returned table includes partial indexes for the fetch CTE and expired-lease reclaim, a partial unique index for timer_id deduplication, and a CHECK constraint ensuring acquired_token and acquired_at are both NULL or both set.

```python
from sqlalchemy import MetaData
from faststream_outbox import make_outbox_table

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

--------------------------------

### Publish from handler with OutboxBroker and OutboxMessage annotations

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/publisher.md

Use these Annotated shortcuts to publish from inside a handler. The broker and message are resolved via Context() paths, and the broker.publish call passes the session to the outbox.

```python
from faststream_outbox.annotations import OutboxBroker, OutboxMessage


@broker.subscriber("orders")
async def handle(msg: OutboxMessage, broker: OutboxBroker) -> None:
    async with session_factory() as session, session.begin():
        await broker.publish({"chained": True}, queue="downstream", session=session)
```

--------------------------------

### tables.py - create outbox table

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/messaging-service.md

Defines the SQLAlchemy Base and creates the outbox table on the app's own MetaData via make_outbox_table, so Alembic owns its migrations.

```python
from faststream_outbox import make_outbox_table
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


OUTBOX_TABLE = make_outbox_table(Base.metadata, table_name="outbox")
```

--------------------------------

### Test publisher with TestOutboxBroker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/testing.md

Tests a publisher by using TestOutboxBroker with a fake producer. The publisher path requires a session argument that must be an AsyncSession, even if mocked. The test verifies that the published message is received by the subscriber.

```python
from unittest.mock import AsyncMock

from sqlalchemy.ext.asyncio import AsyncSession


async def test_publisher() -> None:
    metadata = MetaData()
    outbox_table = make_outbox_table(metadata, table_name="outbox")
    broker = OutboxBroker(None, outbox_table=outbox_table)
    received: list[dict] = []

    @broker.subscriber("orders")
    async def handle(body: dict) -> None:
        received.append(body)

    pub = broker.publisher("orders")

    async with TestOutboxBroker(broker):
        # The publisher path builds its publish command (which validates the
        # session type) before the fake producer is reached, so session= is
        # required and must be an AsyncSession. The fake never touches it — a
        # spec'd mock satisfies the check.
        await pub.publish({"order_id": 1}, session=AsyncMock(spec=AsyncSession))

    assert received == [{"order_id": 1}]
```

--------------------------------

### Add KafkaBroker and publisher to app.py

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/add-kafka-relay.md

Edits to app.py that add a KafkaBroker and a publisher for the orders.kafka topic. The existing broker is renamed to broker_outbox, and broker_kafka.connect is hooked into FastStream's on_startup so the Kafka client opens before the first row is dispatched.

```python
from faststream.kafka import KafkaBroker

broker_outbox = OutboxBroker(engine, outbox_table=outbox_table)
broker_kafka = KafkaBroker("localhost:9092")
kafka_publisher = broker_kafka.publisher("orders.kafka")

app = FastStream(broker_outbox, on_startup=[broker_kafka.connect])
```

--------------------------------

### Architecture diagram

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/messaging-service.md

ASCII diagram showing the architecture: use cases write outbox rows in one transaction, subscribers relay them to Kafka, and ReadMessageUseCase cancels a timer.

```text
  CreateMessageUseCase  ─┐
   (one DB transaction)  ├─▶ outbox row: chat-events    ─▶ subscriber ─▶ Kafka
                         └─▶ outbox row: unread-timers   ─▶ subscriber ─▶ Kafka
                                  ▲
  ReadMessageUseCase ── cancel_timer("unread-timers", timer_id) ┘
```

--------------------------------

### Configure OutboxBroker with OpenTelemetry Prometheus metrics

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/setup-prometheus-opentelemetry.md

Sets up an OutboxBroker with OpenTelemetry metrics exported to Prometheus via a PrometheusMetricReader. The app exposes a /metrics endpoint using make_asgi_app. The adapter is meter-only; spans are not supported. Instrument names match faststream.opentelemetry.TelemetryMiddleware for bus-scope metrics, plus outbox-specific counters.

```python
# app.py — run with `uvicorn app:app --host 0.0.0.0 --port 8000`
from faststream.asgi import AsgiFastStream
from opentelemetry import metrics
from opentelemetry.exporter.prometheus import PrometheusMetricReader
from opentelemetry.sdk.metrics import MeterProvider
from prometheus_client import REGISTRY, make_asgi_app
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine

from faststream_outbox import OutboxBroker, make_outbox_table
from faststream_outbox.metrics.opentelemetry import OpenTelemetryRecorder


# OTel meters → Prometheus reader (scraped at /metrics below)
prometheus_reader = PrometheusMetricReader()
meter_provider = MeterProvider(metric_readers=[prometheus_reader])
metrics.set_meter_provider(meter_provider)

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,
    metrics_recorder=OpenTelemetryRecorder(meter_provider=meter_provider),
)


@broker.subscriber("orders", max_workers=4)
async def handle_order(body: dict) -> None: ...


app = AsgiFastStream(broker, asgi_routes=[("/metrics", make_asgi_app(registry=REGISTRY))])
```

--------------------------------

### Configure subscribers with different lease_ttl_seconds for slow and fast queues

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/subscriber.md

Use this snippet to set up two subscribers: one for slow work with a longer lease TTL (600 seconds) and one for fast work with a shorter TTL (30 seconds). The slow queue tolerates outliers without slowing reclaim of stuck rows elsewhere, while the fast queue keeps reclaim fast. Ensure the TTL is strictly greater than the subscriber's P99 handler duration, accounting for clock skew and queue depth.

```python
@broker.subscriber("slow_q", lease_ttl_seconds=600)  # 10 minutes
async def heavy_job(msg): ...


@broker.subscriber("fast_q", lease_ttl_seconds=30)
async def quick_job(msg): ...
```

--------------------------------

### ioc.py - wire OutboxBroker with modern-di

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/messaging-service.md

Wires the OutboxBroker and engine with modern-di. The caller owns the engine; the broker never disposes it.

```python
from modern_di import Group, Scope, providers
from sqlalchemy.ext.asyncio import create_async_engine
from faststream_outbox import OutboxBroker

from app.tables import OUTBOX_TABLE


class Resources(Group):
    # The caller owns the engine; the broker never disposes it.
    database_engine = providers.Factory(
        scope=Scope.APP,
        creator=lambda: create_async_engine("postgresql+asyncpg://app:app@localhost/app"),
    )
    outbox_broker = providers.Factory(
        scope=Scope.APP,
        creator=lambda engine: OutboxBroker(engine, outbox_table=OUTBOX_TABLE),
        kwargs={"engine": Resources.database_engine},
    )
```

--------------------------------

### Inspect outbox table with psql

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/tutorials/first-outbox-app.md

Verifies the outbox table exists by inspecting its structure in the running PostgreSQL container.

```bash
docker exec outbox-postgres psql -U outbox -d outbox -c '\d outbox'
```

--------------------------------

### OpenTelemetryRecorder

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/observability.md

Bundled adapter for OpenTelemetry observability, equivalent to the Prometheus recorder.

```APIDOC
## OpenTelemetryRecorder

### Description
OTel equivalent of the Prometheus recorder, emitting the same event catalog via the OpenTelemetry API.

### Import
`from faststream_outbox.metrics.opentelemetry import OpenTelemetryRecorder`

### Usage
Pass it to the `metrics_recorder` parameter of `OutboxBroker`:
```python
broker = OutboxBroker(
    engine,
    outbox_table=outbox_table,
    metrics_recorder=OpenTelemetryRecorder(...),
)
```
Full wiring is described in [Setup Prometheus and OpenTelemetry](../setup-prometheus-opentelemetry.md).

```

--------------------------------

### Include an OutboxRouter in the broker

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/router.md

Attaches an OutboxRouter to an OutboxBroker via include_router and creates a FastStream app. The router's subscribers become part of the broker.

```python
from faststream import FastStream
from faststream_outbox import OutboxBroker

from myapp.routers import router

broker = OutboxBroker(engine, outbox_table=outbox_table)
broker.include_router(router)
app = FastStream(broker)
```

--------------------------------

### No prefix: use literal queue name

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/router.md

Shows that OutboxRouter does not accept a prefix argument, so queue names must be fully qualified. Use a literal queue name like "checkout.orders" to namespace queues.

```python
@router.subscriber("checkout.orders")
async def handle_order(...): ...
```

--------------------------------

### Walk every subscriber with broker.subscribers

Source: https://github.com/modern-python/faststream-outbox/blob/main/docs/usage/router.md

Walks every subscriber on the broker, including those attached via include_router. The property iterates over both inline and router-attached subscribers, unlike broker._subscribers which only holds inline ones.

```python
for sub in broker.subscribers:
    ...
```
Tokens:
Raw