Skip to content

Alpha

faststream-sqlbroker is currently in alpha.

Tutorial#

Motivation#

The primary benefit of a message queue built on top of a relational database is the ability to insert messages transactionally, atomically with other database operations, thus enabling the transactional outbox pattern. Also, the relational database is usually the most readily available, already-provisioned piece of infrastructure for a given service.

Given a proper understanding of the trade-offs involved, a relational-database-based queue is an appropriate tool for many low-to-medium throughput, latency-tolerant uses, including as part of a larger messaging flow that involves a "proper" queue (e.g. as an outbox between a service and a queue).

Installation#

PostgreSQL, MySQL, and SQLite are currently supported.

pip install "faststream-sqlbroker"

Schema Variants#

A schema variant selects the consumption semantics the broker implements.

COMPETING_CONSUMERS#

The competing consumers pattern: multiple processes share one queue, each message is handled by one concurrent worker, and processing order is not guaranteed.

Database Tables#

The COMPETING_CONSUMERS variant (version 1) uses up to two tables — message (active messages) and message_archive (completed/failed messages). These settings are grouped under the broker's schema parameter via SqlBrokerSchemaConfig. Set message_archive_table_name=None there to omit using archiving on success and DLQ. You can customize the tables to your liking (partition them, add indices, specify more specific data types like JSONB, etc.) as long as they generally conform to the selected schema definition. Schema check is done on startup if the broker's validate_schema_on_start is True.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
from datetime import datetime, timezone

from sqlalchemy import (
    JSON,
    BigInteger,
    Column,
    DateTime,
    Enum,
    LargeBinary,
    MetaData,
    String,
    Table,
)

from faststream_sqlbroker.sqlbroker.message import SqlBrokerMessageState

metadata = MetaData()

message = Table(
    "message",
    metadata,
    Column("id", BigInteger, primary_key=True),
    Column("queue", String(255), nullable=False, index=True),
    Column("headers", JSON, nullable=True),
    Column("payload", LargeBinary, nullable=False),
    Column(
        "state",
        Enum(SqlBrokerMessageState),
        nullable=False,
        index=True,
        server_default=SqlBrokerMessageState.PENDING.name,
    ),
    Column("attempts_count", BigInteger, nullable=False, default=0),
    Column("deliveries_count", BigInteger, nullable=False, default=0),
    Column(
        "created_at",
        DateTime,
        nullable=False,
        default=lambda: datetime.now(timezone.utc).replace(tzinfo=None),
    ),
    Column("first_attempt_at", DateTime),
    Column(
        "next_attempt_at",
        DateTime,
        nullable=False,
        default=lambda: datetime.now(timezone.utc).replace(tzinfo=None),
        index=True,
    ),
    Column("last_attempt_at", DateTime),
    Column("acquired_at", DateTime),
)


message_archive = Table(
    "message_archive",
    metadata,
    Column("id", BigInteger, primary_key=True),
    Column("queue", String(255), nullable=False, index=True),
    Column("headers", JSON, nullable=True),
    Column("payload", LargeBinary, nullable=False),
    Column("state", Enum(SqlBrokerMessageState), nullable=False, index=True),
    Column("attempts_count", BigInteger, nullable=False),
    Column("deliveries_count", BigInteger, nullable=False),
    Column("created_at", DateTime, nullable=False),
    Column("first_attempt_at", DateTime),
    Column("last_attempt_at", DateTime),
    Column(
        "archived_at",
        DateTime,
        nullable=False,
        default=lambda: datetime.now(timezone.utc).replace(tzinfo=None),
    ),
)

Broker#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from sqlalchemy.ext.asyncio import create_async_engine

from faststream_sqlbroker import (
    SqlBroker,
    SqlBrokerSchemaConfig,
)

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
broker = SqlBroker(
    engine=engine,
    schema=SqlBrokerSchemaConfig(
        message_table_name="message",
        message_archive_table_name="message_archive",
    ),
)

Broker parameters#

  • engine — SQLAlchemy AsyncEngine to use for requests to the database.
  • schema (default: SqlBrokerSchemaConfig()) — SqlBrokerSchemaConfig describing the broker tables and schema selection.
  • schema.message_table_name (default: message) — Name of the table containing active messages.
  • schema.message_archive_table_name (default: message_archive) — Name of the table containing completed/failed messages. Set to None to run without an archive table, in which case subscribers must set both retain_in_archive_on_ack and retain_in_archive_on_reject to False.
  • schema.variant (default: COMPETING_CONSUMERS) — Schema variant to use.
  • schema.version (default: V1) — Variant-specific schema version enum. For COMPETING_CONSUMERS, use SqlBrokerCompetingConsumersSchemaVersion.
  • validate_schema_on_start (default: True) — Validates that the configured tables exist and conform to the expected schema.
  • graceful_timeout (default: 15.0) — Seconds to wait for in-flight messages to finish processing during shutdown.

Publishing#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from datetime import datetime, timedelta, timezone

from sqlalchemy.ext.asyncio import create_async_engine

from faststream import FastStream
from faststream_sqlbroker import SqlBroker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
broker = SqlBroker(engine=engine)
app = FastStream(broker)

publisher_sqlbroker = broker.publisher()

@app.after_startup
async def publish_examples():
    await publisher_sqlbroker.publish("Hello, SqlBroker!", queue="my_queue")

The broker's and publisher's (see publishing) .publish() methods accept:

  • message — The message body.
  • queue (default: "") — The target queue name.
  • headers (default: None) — Optional dict[str, str] of message headers.
  • next_attempt_at (default: None) — Optional datetime (with timezone) for delayed delivery.
  • connection (default: None) — Optional SQLAlchemy AsyncConnection for transactional publishing.

Delayed delivery#

If next_attempt_at is provided, the message won't be fetched until that time. 

1
2
3
4
5
    await publisher_sqlbroker.publish(
        "Process me later",
        queue="my_queue",
        next_attempt_at=datetime.now(timezone.utc) + timedelta(minutes=5),
    )

Transactional publishing#

When connection is provided, the message insert participates in the same database transaction as your other operations, enabling the transactional outbox pattern. 

1
2
3
4
5
6
7
    async with engine.begin() as connection:
        # ... your other database operations using `connection` ...
        await publisher_sqlbroker.publish(
            "Transactional message",
            queue="my_queue",
            connection=connection,
        )

Batch publishing#

The broker's and publisher's .publish_batch() methods insert all messages in a single SQL statement. They accept the same arguments as .publish(), applied to every message in the batch. Wrap an individual payload in SqlBrokerPublishMessage to override its queue, headers, or next_attempt_at.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
    await broker.publish_batch(
        "Hello, SqlBroker!",
        "Another message",
        queue="my_queue",
    )

    await publisher_sqlbroker.publish_batch(
        "Hello, SqlBroker!",
        "Another message",
        queue="my_queue",
    )

    await broker.publish_batch(
        SqlBrokerPublishMessage(
            "Order placed",
            queue="orders",
            headers={"x-source": "checkout"},
            correlation_id="order-1",
        ),
        SqlBrokerPublishMessage(
            "Retry later",
            next_attempt_at=datetime.now(timezone.utc) + timedelta(minutes=5),
        ),
        "Uses batch defaults",
        queue="my_queue",
        headers={"x-default": "batch"},
    )

Subscribing#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from sqlalchemy.ext.asyncio import create_async_engine

from faststream import FastStream

from faststream_sqlbroker import ConstantRetryStrategy
from faststream_sqlbroker import SqlBroker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
broker = SqlBroker(engine=engine)
app = FastStream(broker)


@broker.subscriber(
    queues=["my_queue"],
    retry_strategy=ConstantRetryStrategy(
        delay_seconds=5,
        max_attempts=3,
        max_total_delay_seconds=None,
    ),
    min_fetch_interval=0.1,
    max_fetch_interval=1,
    fetch_batch_size=10,
    flush_interval=1,
)
async def handler(msg: str):
    print(msg)

Subscriber parameters#

  • queues — List of queue names to consume from.
  • max_workers (default: 1) — Number of concurrent handler coroutines.
  • retry_strategy (default: NoRetryStrategy()) — Called to determine if and how soon a Nacked message is retried.
  • fetch_batch_size — Maximum number of messages to fetch in a single batch. A fetch's actual limit might be lower if either the acquired-but-not-yet-processed or acquired-but-not-yet-persisted set has less free capacity.
  • max_not_processed_factor (default: 1.5) — Multiplier for fetch_batch_size to cap the size of the set of acquired-but-not-yet-processed messages.
  • max_not_persisted_factor (default: 2.0) — Multiplier for fetch_batch_size to cap the size of the set of acquired messages whose state has not yet been persisted to the database.
  • min_fetch_interval — Minimum interval between consecutive fetches. If the last fetch was full (returned as many messages as the fetch's limit), the next fetch happens after both (i) minimum fetch interval has passed, and (ii) capacity equal to the fetch batch size has freed up in both the acquired-but-not-yet-processed and acquired-but-not-yet-persisted sets.
  • max_fetch_interval — Maximum interval between consecutive fetches.
  • flush_interval — Interval between flushes of processed message state to the database.
  • release_stuck_interval (default: 60) — Interval between checks for stuck PROCESSING messages.
  • release_stuck_timeout (default: 60 * 10) — Interval since acquired_at after which a PROCESSING message is considered stuck and is released back to PENDING.
  • max_deliveries (default: None) — Maximum number of deliveries allowed for a message for poison message protection. If set, messages that have reached this limit are Rejected without processing. Note that this might violate at-least-once processing semantics.
  • ack_policy (default: REJECT_ON_ERROR) — AckPolicy that controls acknowledgement behavior.
  • retain_in_archive_on_ack (default: True) — Acked messages, in addition to being removed from the primary table, are also persisted in the archive table. Requires the broker to define an archive table (message_archive_table_name).
  • retain_in_archive_on_reject (default: True) — Rejected messages, in addition to being removed from the primary table, are also persisted in the archive table, where they serve as a dead-letter queue. Requires the broker to define an archive table (message_archive_table_name).

Message Lifecycle#

A published message starts out in the message table with a PENDING status. Once a subscriber acquires it, the row is marked as PROCESSING and the message is processed.

From there, the following acknowledgement outcomes are possible:

Ack#

The message is marked as COMPLETED and is moved from the message table to the message_archive table.

Nack#

The retry_strategy is called to determine if the message is allowed to be retried and when it will be retried. If allowed to be retried, the message is marked as RETRYABLE in the message table. If not, the message is Rejected.

Reject#

The message is marked as FAILED and is moved from the message table to the message_archive table.

These outcomes are applied through acknowledgement after a processing attempt. The exception is max_deliveries: if a subscriber sets it and a message exceeds the limit, the message is Rejected before any processing attempt.

Acknowledgements#

Each message's outcome is applied either automatically — driven by the ack_policy — or manually from within the handler.

Automatic via AckPolicy#

Set ack_policy on the subscriber to control what happens after handler execution depending on whether the handler raised an exception or returned.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from sqlalchemy.ext.asyncio import create_async_engine

from faststream import AckPolicy, FastStream

from faststream_sqlbroker import ConstantRetryStrategy, SqlBroker, SqlBrokerMessage

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
broker = SqlBroker(engine=engine)
app = FastStream(broker)


@broker.subscriber(
    queues=["my_queue"],
    ack_policy=AckPolicy.NACK_ON_ERROR,
    retry_strategy=ConstantRetryStrategy(
        delay_seconds=5,
        max_attempts=3,
        max_total_delay_seconds=None,
    ),
    max_fetch_interval=1.0,
    min_fetch_interval=0.1,
    fetch_batch_size=10,
    flush_interval=1.0,
)
async def automatic_handler(msg: str) -> None:
    print(msg)
  • AckPolicy.ACK — Acks the message after the handler attempt, even if the handler raised an exception.
  • AckPolicy.ACK_FIRST — Same as AckPolicy.ACK for this broker.
  • AckPolicy.REJECT_ON_ERROR — On success, Acks the message. On exception, Rejects the message. retry_strategy is ignored.
  • AckPolicy.NACK_ON_ERROR — On success, Acks the message. On exception, Nacks the message. With NoRetryStrategy() or None in retry_strategy, this has the same effect as REJECT_ON_ERROR.
  • AckPolicy.MANUAL — Requires explicit msg.ack(), msg.nack(), or msg.reject() in the handler. In the absence of explicit action, the message is Rejected as a safety precaution.

Automatic acknowledgement applies only if the handler did not already call one of the manual acknowledgement methods.

Manual#

Use AckPolicy.MANUAL when the handler should decide the outcome explicitly with msg.ack(), msg.nack(), or msg.reject().

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
@broker.subscriber(
    queues=["my_queue"],
    ack_policy=AckPolicy.MANUAL,
    max_fetch_interval=1.0,
    min_fetch_interval=0.1,
    fetch_batch_size=10,
    flush_interval=1.0,
)
async def manual_handler(msg: SqlBrokerMessage, body: str) -> None:
    await msg.ack()
    print(body)

Manual acknowledgements can also be used with any other ack_policy, not just AckPolicy.MANUAL. They override the ack_policy driven action.

Retry strategies#

When a message is Nacked (either manually with msg.nack() or by AckPolicy.NACK_ON_ERROR), the retry_strategy determines if and when the message should be retried. By default, NoRetryStrategy() disables retries. All strategies accept common parameters:

  • max_attempts - Maximum number of processing attempts.
  • max_total_delay_seconds - Maximum delay between the first and last attempt.

If either limit is reached, the message is marked as Rejected. Otherwise, next_attempt_at is set on the message to signify a scheduled retry.

ConstantRetryStrategy#

Retries after a fixed delay_seconds every time.

1
2
3
4
5
constant = ConstantRetryStrategy(
    delay_seconds=5,
    max_attempts=3,
    max_total_delay_seconds=None,
)

LinearRetryStrategy#

First retry after initial_delay_seconds, then the delay increases by step_seconds with each attempt.

1
2
3
4
5
6
linear = LinearRetryStrategy(
    initial_delay_seconds=1,
    step_seconds=2,
    max_attempts=3,
    max_total_delay_seconds=60,
)

ExponentialBackoffRetryStrategy#

Delay starts at initial_delay_seconds and is multiplied by multiplier on each attempt. max_delay_seconds caps the delay.

1
2
3
4
5
6
7
exponential = ExponentialBackoffRetryStrategy(
    initial_delay_seconds=1,
    multiplier=2.0,  # default
    max_delay_seconds=60,
    max_attempts=3,
    max_total_delay_seconds=300,
)

ExponentialBackoffWithJitterRetryStrategy#

Same as exponential backoff, but adds random jitter (up to delay * jitter_factor) to spread out retries and avoid thundering herds.

1
2
3
4
5
6
7
8
exponential_jitter = ExponentialBackoffWithJitterRetryStrategy(
    initial_delay_seconds=1,
    multiplier=2.0,  # default
    max_delay_seconds=60,
    jitter_factor=0.5,  # default
    max_attempts=3,
    max_total_delay_seconds=300,
)

ConstantWithJitterRetryStrategy#

Retries after base_delay_seconds plus random jitter in the range [-jitter_seconds, +jitter_seconds].

1
2
3
4
5
6
constant_jitter = ConstantWithJitterRetryStrategy(
    base_delay_seconds=5,
    jitter_seconds=2,
    max_attempts=3,
    max_total_delay_seconds=None,
)

NoRetryStrategy#

No retries — the message is marked as Rejected on the first Nack.

1
2
3
no_retry = NoRetryStrategy(
    max_attempts=1,  # default
)

Transactional outbox#

Implementing the transactional outbox pattern becomes as simple as the following.

Publish messages transactionally with your other database operations. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from sqlalchemy.ext.asyncio import create_async_engine

from faststream import FastStream
from faststream.kafka import KafkaBroker, KafkaPublishMessage

from faststream_sqlbroker import SqlBroker, SqlBrokerMessage

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
broker_sqlbroker = SqlBroker(engine=engine)
broker_kafka = KafkaBroker("127.0.0.1:9092")
app = FastStream(broker_sqlbroker, on_startup=[broker_kafka.connect])
publisher_sqlbroker = broker_sqlbroker.publisher()


@app.after_startup # just an example
async def publish_examples():
    async with engine.begin() as connection:
        # ... your other database operations using `connection` ...
        await publisher_sqlbroker.publish(
            {"message": "Hello, SqlBroker!"},
            queue="sqlbroker_queue",
            headers={
                "x-test-header": "outbox",
                "x-kafka-key-source": "outbox-key",
            },
            connection=connection,
        )

And relay the messages from the database to another broker. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
publisher_kafka = broker_kafka.publisher("kafka_topic")


@publisher_kafka
@broker_sqlbroker.subscriber(
    queues=["sqlbroker_queue"],
    max_fetch_interval=1,
    min_fetch_interval=0,
    fetch_batch_size=10,
    flush_interval=3,
)
async def handle_msg(
    msg_body: dict,
    msg: SqlBrokerMessage,
) -> KafkaPublishMessage:
    return KafkaPublishMessage(
        msg_body,
        headers={
            "x-test-header": msg.headers["x-test-header"],
        },
        key=msg.headers["x-kafka-key-source"].encode(),
    )

Observability#

FastStream already supplies Prometheus metrics for message publishing and processing rates and latencies through its Prometheus middleware.

Grafana panels from the FastStream Prometheus middleware: publish and process rates, publish and process duration percentiles, messages in process, and received message size

SQLBroker additionally provides metrics derived from the messages persisted in the database:

  • sqlbroker_messages — messages in the primary table, labeled by queue and state.
  • sqlbroker_most_overdue_message_age_seconds — how long the most overdue message has been eligible for processing, labeled by queue and state.
  • sqlbroker_archived_messages — messages in the archive table, labeled by queue and state.
  • sqlbroker_state_collection_last_success_timestamp_seconds — Unix timestamp of the last successful database sample.
Grafana panels from the SQLBroker state sampler: messages by queue and state, oldest message age, and archived messages by queue and state

Standalone sampler#

If the sampler runs in every broker node, each node queries the shared database and reports database-wide values and exports a duplicate copy of the same series. Prefer one standalone sampler per database, using the packaged sqlbroker-state-metrics command:

pip install "faststream-sqlbroker[cli]"
sqlbroker-state-metrics \
    --host 0.0.0.0 \
    --port 8000 \
    --message-table message \
    --archive-table message_archive \
    --interval 30 \
    --database-url postgresql+asyncpg://user:pass@localhost/mydb # pragma: allowlist secret

In-broker sampler#

The sampler can also run as part of the broker. Install the Prometheus dependency and pass the registry exposed by your metrics endpoint to SqlBrokerStateMetricsConfig:

pip install "faststream-sqlbroker[prometheus]"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from prometheus_client import CollectorRegistry, make_asgi_app
from sqlalchemy.ext.asyncio import create_async_engine

from faststream_sqlbroker import SqlBroker
from faststream_sqlbroker.sqlbroker.observability import (
    SqlBrokerStateMetricsConfig,
)

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
registry = CollectorRegistry()

broker = SqlBroker(
    engine=engine,
    state_metrics_config=SqlBrokerStateMetricsConfig(
        registry=registry,
        interval=30,
    ),
)
metrics_app = make_asgi_app(registry=registry)

Mount metrics_app at /metrics in your ASGI application. These database-wide gauges must not be summed across instances. This applies even to nominally single-node deployments because rolling restarts can briefly run the old and new broker nodes at the same time.