faststream-outbox is a FastStream broker integration for the transactional outbox pattern, with a Postgres table as the message queue.
A producer writes a domain entity and an outbox row in the same SQLAlchemy transaction by calling broker.publish(body, queue=..., session=session). A separate subscriber polls the table and relays each row to a real message bus (Kafka, RabbitMQ, NATS, Redis…) with a single decorator, or processes the rows in place if you don't have a downstream broker.
Quickstart: outbox relay to Kafka
Write the outbox row in your domain transaction; relay rows to Kafka with a stacked decorator.
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://localhost/app")
broker_outbox = OutboxBroker(engine, outbox_table=outbox_table)
broker_kafka = KafkaBroker("127.0.0.1:9092")
publisher_kafka = broker_kafka.publisher("orders")
@publisher_kafka
@broker_outbox.subscriber("orders_outbox")
async def relay(body: dict) -> dict:
return body
app = FastStream(broker_outbox, on_startup=[broker_kafka.connect])
# Producer side — share the caller's open transaction; on commit both the
# domain row and the outbox row land atomically. The relay subscriber picks
# the outbox row up and publishes it to Kafka with at-least-once delivery.
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session, session.begin():
session.add(Order(id=1))
await broker_outbox.publish(
{"order_id": 1},
queue="orders_outbox",
session=session,
)
The same one-decorator pattern works for RabbitMQ, NATS, Redis, and Confluent. See the relay tutorial for the FastAPI lifecycle, header propagation, router shapes, and the at-least-once contract.
Quickstart: standalone outbox queue
If you don't have a downstream broker, the same broker can process outbox rows in place, and the table is the queue.
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://localhost/app")
broker = OutboxBroker(engine, outbox_table=outbox_table)
app = FastStream(broker)
@broker.subscriber("orders", max_workers=4)
async def handle(order_id: int) -> None:
print(f"order {order_id}")
# Producer side — share the caller's open transaction:
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session, session.begin():
session.add(Order(id=1))
await broker.publish(1, queue="orders", session=session)
How it works
A subscriber owns two async loops: a fetch loop claims available rows via a single CTE (SELECT … FOR UPDATE SKIP LOCKED → UPDATE acquired_token=:uuid, acquired_at=now() RETURNING *), and max_workers worker loops dispatch to the handler. On success, DELETE WHERE id=:id AND acquired_token=:token; on failure, the retry strategy schedules another attempt or terminally drops the row. Terminal failures DELETE by default; pass dlq_table=make_dlq_table(metadata) to atomically archive them into a sibling audit table instead (see Dead-letter queue).
The acquired_token is the load-bearing invariant. If a slow handler's lease expired and another worker re-claimed the row, the slow handler's terminal DELETE is a no-op because the token no longer matches, so it cannot clobber the new lease holder.
With the asyncpg driver, the fetch loop also LISTENs on outbox_<table> and publish emits pg_notify(...), so idle dispatch latency is ~10ms instead of up to max_fetch_interval.
See How it works for the full architecture.
Optional extras
faststream-outbox[asyncpg]: asyncpg driver (enablesLISTEN/NOTIFYfor ~10ms idle dispatch)faststream-outbox[fastapi]: FastAPI integration viafaststream_outbox.fastapi.OutboxRouterfaststream-outbox[validate]: Alembic forbroker.validate_schema()faststream-outbox[prometheus]: Prometheus metrics adapterfaststream-outbox[opentelemetry]: OpenTelemetry metrics adapter
Acknowledgements
The architecture of this package is heavily informed by Arseniy Popov's PR #2704 (feat: add sqla broker) on upstream FastStream. The FastStream broker/registrator/subscriber wiring, the SELECT … FOR UPDATE SKIP LOCKED fetch-and-claim CTE, the retry strategy hierarchy, and the in-transaction publish contract all originate from there. This package is a Postgres-only reimplementation. It differs in storage model (lease tokens in place of an explicit state column, and an opt-in archive table), loop structure (two loops where the PR has four), and wake-up mechanism (LISTEN/NOTIFY), and it adds timer mechanics. Credit for the original design belongs to Arseniy.
Links
Part of modern-python
Browse the full list of templates and libraries in
modern-python; the org profile has the categorized index.
Metadata
Release files for faststream-outbox 0.18.2
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| faststream_outbox-0.18.2.tar.gz | 96.1 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| faststream_outbox-0.18.2-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 211.4 kB
Release files / faststream_outbox-0.18.2.tar.gz
| Download URL | faststream_outbox-0.18.2.tar.gz |
|---|---|
| Size | 96.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
a8dd501ec44d2878d50bd2506c58524a70713f0a69e0bf1a570ec44e090e9126
|
|
BLAKE2b-256 checksum How to use checksums |
0966fc3feb59a40080282e3290999a22d90c17529921d532c27c2556e902aabd
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.5 {"installer":{"name":"uv","version":"0.13.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / faststream_outbox-0.18.2-py3-none-any.whl
| Download URL | faststream_outbox-0.18.2-py3-none-any.whl |
|---|---|
| Size | 115.3 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
cd92ed83da340564897cae0b9a748669df1ccf8954ff7ce85338b13455241df9
|
|
BLAKE2b-256 checksum How to use checksums |
487a1b181360aeddba30e17fbfe66ddcbb0910f7d403c41ba4ae2996826a689f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.5 {"installer":{"name":"uv","version":"0.13.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Security for faststream-outbox 0.18.2
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
