Schema validation

The package never creates or migrates your schema (that's Alembic's job), but it does provide an opt-in helper that verifies the live table has everything the broker needs at runtime.

broker.validate_schema() delegates to Alembic's autogenerate.compare_metadata against a throwaway MetaData populated by make_outbox_table(...). The canonical Table is the single source of truth; the validator never duplicates the schema declaration. When the broker was constructed with a dlq_table, validate_schema() runs a second pass over the DLQ table the same way.

Alembic's comparison cannot see partial-index predicates or check constraints, so the validator also queries pg_catalog (pg_index, pg_constraint) directly and reports a missing or drifted partial-index predicate or lease check constraint.

Install

Alembic is an optional dependency:

pip install 'faststream-outbox[validate]'

Calling validate_schema() without this extra raises ImportError. Every other code path (producers, subscribers, retries, timers, LISTEN/NOTIFY) works without it.

Usage

await broker.validate_schema()

Raises RuntimeError if the live table is missing what the broker needs: absent table, missing columns, mismatched column types, flipped nullability, missing partial indexes.

Pass check_autovacuum=True to also check that the outbox table carries the recommended autovacuum reloptions (see Alembic migrations, outbox_autovacuum_ddl). An untuned table raises a RuntimeError whose message starts with Outbox autovacuum not tuned:, separate from any schema mismatch. The default (False) skips this check.

await broker.validate_schema(check_autovacuum=True)

Extras are intentionally ignored: the validator only flags missing schema (add_* / modify_* ops). remove_* ops are silently dropped so you can attach your own audit columns or additional indexes without the validator complaining.

Some drift cannot be fixed by re-running alembic revision --autogenerate: a missing/altered outbox_lease_ck CHECK or a drifted partial-index predicate. For those, the RuntimeError ends with a pointer to Alembic migrations § Fixing drift autogenerate can't see, which holds the hand-written migration recipe.

Two indexes changed shape between releases, so a table migrated on an older version fails validation after an upgrade until it gets the new index. For those, the error names the release and links to the upgrade migration:

Outbox schema mismatch: table 'outbox' missing index over ['queue', 'next_attempt_at', 'id'] (name='outbox_leased_idx', unique=False); a table created before 0.17 needs it in place of outbox_lease_idx: https://faststream-outbox.modern-python.org/operations/alembic/#upgrading-to-017

Missing index	Changed in	Upgrade migration
outbox_pending_idx	0.15	Upgrading to 0.15
outbox_leased_idx	0.17	Upgrading to 0.17

Use the linked migration, not alembic revision --autogenerate: it builds the index without CONCURRENTLY and blocks writes for the whole build.

Server defaults are not checked

The diff runs with compare_server_default=False. Alembic's server-default comparison is flaky against Postgres' normalized expressions (now() vs CURRENT_TIMESTAMP), so it is disabled to avoid false positives. A green validate_schema() therefore does not prove your server defaults exist. The load-bearing case: a table missing server_default=now() on next_attempt_at leaves fresh rows with NULL next_attempt_at, which the fetch CTE's next_attempt_at <= now() predicate silently filters out. The result is a broker outage that validation will not catch. Generate your migration from make_outbox_table(...) so the defaults are in place to begin with.

Where to call it

Call it from a /health endpoint or startup hook, not at broker.start(). If validate_schema() ran at startup and your migration hadn't been applied yet, the broker would crash-loop itself. Operators need to be able to roll out a new schema version and have Alembic catch up against the same DB without a startup loop.

A typical pattern under FastAPI:

from fastapi import FastAPI


app = FastAPI()


@app.get("/health")
async def health() -> dict:
    await broker.validate_schema()
    return {"ok": True}

Or as a one-shot CI check after running migrations:

import asyncio

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

from faststream_outbox import OutboxBroker, make_outbox_table


async def main() -> None:
    engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost/outbox")
    outbox_table = make_outbox_table(MetaData(), table_name="outbox")
    broker = OutboxBroker(engine, outbox_table=outbox_table)
    await broker.validate_schema()
    print("schema OK")


asyncio.run(main())

CI recipe: Alembic migrations § Drift detection in CI.

In tests

FakeOutboxClient.validate_schema() raises NotImplementedError: there is no real DB to validate against, and a silent pass would let users ship broken schemas while their TestOutboxBroker-backed tests stay green.

Tests that need real schema validation must build an OutboxBroker(engine=real_engine, outbox_table=table) against the same DSN the migrations ran against and call await broker.validate_schema(). See Testing.