Queues and Tasks¶

Background work without the pain. The SDK wraps FastStream (messaging) and TaskIQ (tasks + scheduling) in typed classes with a single vocabulary — you never import faststream or taskiq in application code.

Which tool?

MessageBroker (messaging) — an event happens and many services/consumers react. Fan-out, at-least-once, decoupled from the request. E.g. "order paid" → inventory, email, analytics.
TaskQueue (tasks) — offload slow work from one request handler to a worker, keeping the HTTP response fast. E.g. send an email, render a PDF.
TaskQueue.cron / .interval (scheduling) — periodic runs.
Outbox — when publishing must be atomic with a database INSERT.

Every class shares the same lifecycle — connect() / disconnect() / lifespan() / health_check() / is_connected — and exposes the raw underlying object (.broker) as an escape hatch.

Messaging — `MessageBroker`¶

The problem FastStream handles poorly: its API changes shape with the transport. You subscribe with @broker.subscriber("q") and publish with broker.publish(msg, queue="q") on RabbitMQ, topic= on Kafka, subject= on NATS. Confusing and non-portable.

MessageBroker hides that behind one concept: a channel (a string). You publish to a channel and everyone subscribed to it receives the message.

Install with [queue] (pulls faststream[rabbit]).

# src/queue/__init__.py

from pydantic import BaseModel

from tempest_fastapi_sdk.queue import MessageBroker

from src.core.settings import settings
from src.services.orders import mark_order_paid


# Pick the transport with a constructor — no faststream import.
mq = MessageBroker.rabbitmq(settings.RABBITMQ_URL)


class OrderPaid(BaseModel):
    order_id: str
    user_id: str


class OrderCancelled(BaseModel):
    order_id: str
    reason: str


@mq.on("orders.paid")
async def handle_order_paid(event: OrderPaid) -> None:
    """Receives every event published to the 'orders.paid' channel."""
    await mark_order_paid(event.order_id, event.user_id)

Note the event: OrderPaid: the type hint drives decoding. FastStream validates the inbound payload into that Pydantic model before your handler runs — a malformed message never reaches your code.

Wire the lifecycle into the FastAPI lifespan and publish from anywhere:

# src/api/app.py
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from src.queue import mq, OrderPaid


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    await mq.connect()
    try:
        yield
    finally:
        await mq.disconnect()


app = FastAPI(lifespan=lifespan)


@app.post("/orders/{order_id}/pay")
async def pay_order(order_id: str) -> dict[str, str]:
    """Publish from any service/handler — channel first, message second."""
    await mq.publish("orders.paid", OrderPaid(order_id=order_id, user_id="u1"))
    return {"status": "published"}

Transports

MessageBroker.rabbitmq(url), .redis(url), .kafka(*servers), .nats(servers). Each lazily imports the right FastStream backend and raises with the exact install command if the extra is missing. Need a custom (or test) broker? MessageBroker(my_broker).

Recap

MessageBroker.rabbitmq(url) — pick the transport, hide FastStream.
@mq.on("channel") — declare a consumer; the parameter type validates the message.
await mq.publish("channel", model) — publish; channel first.
mq.publish(...) only works after connect() (raises RuntimeError before).

Wire it into the health router: make_health_router(checks={"queue": mq.health_check}).

Class-based consumers¶

Prefer grouping handlers in a class (shared setup, inheritance) over free functions? Consumer offers two styles, both explicit (nothing is guessed from the class name). Register with mq.register(...).

Constructor form — pass the channel and the Pydantic schema to the constructor; override handle:

from tempest_fastapi_sdk.queue import Consumer

from src.queue import OrderPaid, mq
from src.services.orders import mark_order_paid


class OrderPaidConsumer(Consumer):
    async def handle(self, event: OrderPaid) -> None:
        await mark_order_paid(event.order_id)


mq.register(OrderPaidConsumer(channel="orders.paid", schema=OrderPaid))

Grouped form — one class, many channels, each method marked with @subscribe; the schema is the method's own annotation:

from tempest_fastapi_sdk.queue import Consumer, subscribe

from src.queue import OrderCancelled, OrderPaid, mq


# OrderPaid / OrderCancelled defined in the `src/queue/__init__.py` block above.


class OrdersConsumer(Consumer):
    @subscribe("orders.paid")
    async def on_paid(self, event: OrderPaid) -> None: ...

    @subscribe("orders.cancelled")
    async def on_cancelled(self, event: OrderCancelled) -> None: ...


mq.register(OrdersConsumer())

Explicit, no magic

In the constructor form the schema is passed explicitly in __init__ and is what validates the payload — no annotation-sniffing. In the grouped form the schema is the method's visible annotation. The @mq.on(...) function decorator is still available — pick the style you like.

The channel may be a string or a QueueSpec, exactly as in @mq.on(...) — declaring a dead-letter exchange or a quorum queue does not push you back to the decorator. The same holds for the rest of what @mq.on(...) takes: prefetch= and the transport's own options (exchange=, say) exist in both forms.

from faststream.rabbit import RabbitExchange

from tempest_fastapi_sdk.queue import Consumer, subscribe

from src.queue import OrderCancelled, OrderPaid, mq


class OrdersConsumer(Consumer):
    """One cap for the whole class; one heavy method with its own."""

    prefetch = 32

    @subscribe("reports.generate", prefetch=1)
    async def generate(self, event: OrderPaid) -> None: ...

    @subscribe("orders.cancelled", exchange=RabbitExchange("events", durable=True))
    async def on_cancelled(self, event: OrderCancelled) -> None: ...


mq.register(OrdersConsumer())

Class-based publishers¶

Consumer covered consumption; publishing stayed loose — await mq.publish("orders.paid", event), with the channel as an arbitrary string and the payload typed Any. Nothing connected the two ends of what is, in practice, one contract.

Publisher is that half. It carries the channel and the payload model as class attributes, and publish takes exactly the declared type:

from tempest_fastapi_sdk.queue import Publisher

from src.queue import ORDERS_PAID, OrderPaid, mq


class OrderPaidPublisher(Publisher[OrderPaid]):
    channel = ORDERS_PAID
    schema = OrderPaid


orders = mq.publisher_for(OrderPaidPublisher)


async def confirm_order(order_id: str) -> None:
    """Announce that the order was paid.

    Args:
        order_id (str): The confirmed order.
    """
    await orders.publish(OrderPaid(order_id=order_id))

Three things the loose call could not give you:

The type checker sees the payload. Publisher[OrderPaid] makes publish take an OrderPaid, so publishing the wrong model is a red squiggle in your editor rather than a message the consumer rejects in production.
The schema is enforced on the way out. If the object is not an instance of the declared model, publish raises TypeError — the consumer is a process away and can only reject what already left.
The topology is registered. A QueueSpec on channel goes through the same binding @mq.on(...) uses, so a service that only publishes still declares the dead-letter exchange it names.

Channel and schema need not be class attributes. Both also go through __init__ — useful when the channel is only known at runtime (per tenant, per environment) and you do not want a subclass per value:

from tempest_fastapi_sdk.queue import Publisher

from src.queue import OrderPaid, mq

orders: Publisher[OrderPaid] = Publisher(mq, channel="orders.paid", schema=OrderPaid)

publisher_for takes the same two, and they win over what the class declares:

from tempest_fastapi_sdk.queue import Publisher

from src.queue import OrderPaid, mq


class TenantPublisher(Publisher[OrderPaid]):
    schema = OrderPaid


def publisher_for_tenant(tenant: str) -> Publisher[OrderPaid]:
    """Return the publisher for that tenant's channel.

    Args:
        tenant (str): The tenant identifier.

    Returns:
        The publisher bound to `orders.paid.<tenant>`.
    """
    return mq.publisher_for(TenantPublisher, channel=f"orders.paid.{tenant}")

channel and schema are named parameters, not **options — so the type checker sees them and a publish option sharing one of those names is not swallowed.

Not the same as mq.publisher(channel)

mq.publisher(...) returns FastStream's own publisher object — an escape hatch, useful mainly because it makes the channel show up in the generated AsyncAPI. Publisher goes through mq.publish(), so it keeps the message_id deduplication depends on and the traceparent / x-request-id headers tracing depends on. A publisher that bypassed those would look identical and silently break both.

Queue topology — `QueueSpec`¶

A channel as a string covers most cases. What it does not express is exactly what decides whether the queue survives a restart, where a rejected message goes, and how long it lives. On RabbitMQ that lives in the queue declaration, not in the name.

QueueSpec carries that topology as typed data, and is accepted anywhere the string was:

from pydantic import BaseModel

from tempest_fastapi_sdk.queue import DeadLetterSpec, MessageBroker, QueueSpec, QueueType

mq = MessageBroker.rabbitmq("amqp://guest:guest@localhost:5672/")


class OrderPaid(BaseModel):
    order_id: str
    amount_cents: int

ORDERS_PAID = QueueSpec(
    name="orders.paid",
    dead_letter=DeadLetterSpec(exchange="dlx"),
    message_ttl_ms=60_000,
    queue_type=QueueType.QUORUM,
)


@mq.on(ORDERS_PAID)
async def handle(event: OrderPaid) -> None:
    """Consume from a durable, quorum queue with dead-lettering."""

It translates to the arguments AMQP expects:

{"x-queue-type": "quorum", "x-dead-letter-exchange": "dlx", "x-message-ttl": 60000}

Without dead_letter, a failure is a silent discard

The consumer policy is REJECT_ON_ERROR: a handler that raises issues basic.reject with requeue=False. That avoids a poison-message loop — but without x-dead-letter-exchange RabbitMQ throws the message away. No error, no dead queue, no metric. That is why DeadLetterSpec exists.

The exchange has to exist¶

RabbitMQ happily declares a queue pointing at an x-dead-letter-exchange that does not exist — and then discards at routing time, silently. So connect() declares the exchanges named by the registered QueueSpec, as durable topic exchanges:

from tempest_fastapi_sdk.queue import MessageBroker


async def startup(mq: MessageBroker) -> None:
    """Start the broker; the registered specs' DLX are declared here."""
    await mq.connect()

Where the broker is managed and the application has no permission to declare, turn it off and handle topology outside:

from tempest_fastapi_sdk.queue import MessageBroker

mq = MessageBroker.rabbitmq(
    "amqp://guest:guest@localhost:5672/",
    declare_topology=False,
)

A field the transport cannot express raises¶

MessageBroker is multi-transport, and dead_letter / TTL / priority are AMQP. Asking for them on a broker without the concept is not ignored:

UnsupportedTopologyError: QueueSpec('orders.paid') sets dead_letter, which the
kafka transport cannot express. Remove the field, or use a bare channel name
and configure the topology outside the SDK.

Ignoring it silently would produce a queue that looks configured and discards every failure — exactly the defect QueueSpec exists to prevent. Same choice as op.replace_enum, which raises on an unsupported dialect instead of emitting DDL that does nothing.

A bare QueueSpec(name=...) stays portable on any transport: it asks for nothing beyond the name.

Event-path reliability¶

The consumer policy is REJECT_ON_ERROR. A handler that raises issues basic.reject with requeue=False — no loop, and gone. Three pieces close that, mirroring what TaskQueue already has.

Dead-letter: a failure becomes a record¶

from tempest_fastapi_sdk.queue import MessageBroker
from tempest_fastapi_sdk.tasks import DbDeadLetterSink


def wire_dead_letter(mq: MessageBroker, sink: DbDeadLetterSink) -> None:
    """Send every terminal consumer failure to the sink.

    Args:
        mq (MessageBroker): The broker, before connect().
        sink (DbDeadLetterSink): Where the dead event is stored.
    """
    mq.dead_letter(sink, max_attempts=3)

The sink is the same protocol the task path uses, so DbDeadLetterSink, the admin panel and make_requeue_action work unchanged — a dead task and a dead event on one screen.

The mapping is deliberate: task_name carries the channel, task_id the broker's message id, and kwargs["body"] the raw body.

Reported once, not per attempt

The sink fires only on the delivery that exhausts max_attempts, read from the x-death header. Alerting on every attempt turns one bad message into a stream of alerts.

Delayed retry, performed by the broker¶

AMQP has no per-message delay. The portable way is a pair of queues: the main one sends the rejected message to a queue whose only job is to hold it, and that queue's TTL returns it to the main exchange when it expires.

from tempest_fastapi_sdk.queue import ConsumerRetryPolicy, retry_queues

from tempest_fastapi_sdk.queue import (
    ConsumerRetryPolicy,
    MessageBroker,
    retry_queues,
)

TOPOLOGY = retry_queues(
    "orders.paid",
    ConsumerRetryPolicy(max_attempts=3, delay_ms=30_000),
    retry_exchange="orders.retry",
    main_exchange="orders",
    dead_exchange="orders.dead",
)


async def wire_retry(mq: MessageBroker) -> None:
    """Declare and bind the three queues of the retry chain.

    Args:
        mq (MessageBroker): The broker, already connected.
    """
    await mq.declare_retry_topology(TOPOLOGY)

Declaring without binding discards the message

Declaring the queues is not enough: each has to be bound to its exchange. Without the bindings, a rejected message is routed into an exchange with nothing behind it — and RabbitMQ drops it silently. Measured against a real broker: with the bindings the message comes back on schedule (1.5s gaps for a 1.5s TTL); without them it is delivered once and vanishes. declare_retry_topology() does both.

The broker does the waiting, so a worker restart in the meantime changes nothing. The alternative is the rabbitmq_delayed_message_exchange plugin, simpler to declare and requiring the plugin — unavailable on several managed offerings, including the free CloudAMQP tier.

The topology alone retries forever

AMQP counts redeliveries in x-death but will not stop on its own. What enforces max_attempts is the dead_letter() middleware. Declaring the topology without installing the middleware yields infinite retries — which is why they are documented together.

How x-death is read (and why the expired entry does not count)

RabbitMQ keeps one entry per (queue, reason) pair, and the retry chain dead-letters twice per round: rejected out of the main queue, and expired out of the waiting queue when its TTL fires. Summing every entry advances the counter by two per retry — with max_attempts=3 the message was dropped after two runs. delivery_attempt() counts only failed deliveries, skipping expired.

Metrics¶

from tempest_fastapi_sdk.queue import MessageBroker, QueueMetrics


def wire_metrics(mq: MessageBroker) -> None:
    """Publish consume counts and durations on the shared /metrics.

    Args:
        mq (MessageBroker): The broker, before connect().
    """
    mq.enable_metrics(QueueMetrics())

Produces queue_messages_total{channel,status} and queue_message_duration_seconds{channel}. Without it the consumer failure rate is invisible — the message is rejected, the broker discards it, and nothing counts.

status is ok, error or duplicate. The last is a delivery deduplicate() rejected because a sibling worker held the claim: its own label, precisely so that deduplication doing its job does not inflate the error rate an alert is built on.

Prefetch — how many messages stay in flight¶

Uncapped, the broker delivers as fast as the consumer acks. Three consequences, all in production: a slow handler accumulates messages in process memory; the first replica to connect takes the batch and its siblings sit idle; and the unacked backlog is held in RAM until the pod is OOM-killed and the whole lot is redelivered.

from tempest_fastapi_sdk.queue import MessageBroker

mq = MessageBroker.rabbitmq("amqp://guest:guest@localhost:5672/", prefetch=32)


@mq.on("reports.generate", prefetch=1)
async def generate(request: dict[str, str]) -> None:
    """Heavy handler: a low cap, without throttling its neighbours."""

The broker value applies to the connection; the consumer value overrides it for that consumer alone.

On the class path the knob is the same one, at three heights — constructor, class and method:

from tempest_fastapi_sdk.queue import Consumer, subscribe

from src.queue import OrderPaid, mq


class ReportsConsumer(Consumer):
    prefetch = 32                       # every binding this class declares

    @subscribe("reports.generate", prefetch=1)
    async def generate(self, event: OrderPaid) -> None:
        """Heavy handler: the method's cap beats the class's."""


class OrderPaidConsumer(Consumer):
    async def handle(self, message: OrderPaid) -> None: ...


mq.register(ReportsConsumer())
mq.register(OrderPaidConsumer(channel="orders.paid", schema=OrderPaid, prefetch=8))

Why prefetch is a named parameter rather than one more **options

FastStream has no prefetch keyword — it carries basic.qos on a Channel object. Forwarding the raw word raises TypeError: RabbitRegistrator.subscriber() got an unexpected keyword argument 'prefetch', which is exactly what the class path did until v0.209.0. Naming the parameter is what allows the translation — and what the type checker sees.

There is no good default I could guess

Current behaviour is uncapped, and this PR does not change that — it exposes the knob. Too small serializes consumption and destroys throughput; too large recreates the problem. The right number depends on your handler's latency, and fixing one without measuring would repeat the mistake DEFAULT_INTRA_OP_THREADS made in modelops before it was re-justified. Measure with a known-latency consumer before choosing.

Prefetch is not handler concurrency

Prefetch caps how many messages the broker delivers unacked. How many coroutines run at once is a separate decision. Confusing the two is common: prefetch=1 does not serialize the handler if the handler itself fans out.

Publisher confirms are already on

FastStream's Channel defaults to publisher_confirms=True, so a publish lost to a broker restart is not silent. Pinned by a test.

Consume-side idempotency¶

Delivery is at-least-once: worker restart, nack with requeue, an ack lost to the network. Redelivery is not a rare case, it is the normal mode.

from tempest_fastapi_sdk.queue import MessageBroker, RedisDedupStore


def wire_dedup(mq: MessageBroker, redis: object) -> None:
    """Run each message id at most once.

    Args:
        mq (MessageBroker): The broker, before connect().
        redis (object): An async Redis client.
    """
    mq.deduplicate(RedisDedupStore(redis), ttl_seconds=86_400)

publish() now generates a message_id when you do not pass one — without a stable id there is no key to deduplicate on, and a redelivery is indistinguishable from a new event.

Two-phase marking: the first delivery marks in_flight and runs; success marks done and the next one is skipped; a failure releases the key, so a retry actually retries. Without that third part, trading "processed twice" for "processed never" would be worse than the original problem.

This is not exactly-once — nothing here is

The mark and the handler's effect are not atomic. A crash between them leaves an in_flight key that expires, and the message runs again. This is at-least-once with a much smaller window, not exactly-once.

When the database already solves it, use the database

If the handler's effect is a row keyed by something the domain owns, INSERT ... ON CONFLICT DO NOTHING is idempotent with no extra moving part — no TTL to tune, no second store to operate. This middleware is for effects that are not rows: an email, a third-party call, a downstream publish.

A concurrent delivery is rejected, not acknowledged

If another worker holds the claim, this copy raises ConcurrentDeliveryError and the broker rejects it. Acknowledging would be dangerous: the in-flight worker may still fail, and the copy that could have retried would be gone.

Tracing and request id across the queue¶

The request opens a trace, publishes an event and answers 201 — and the consumer that charges the card, writes the ledger and sends the mail showed up as three orphan traces, with no parent and no relation to each other.

from tempest_fastapi_sdk.queue import MessageBroker


def wire_tracing(mq: MessageBroker) -> None:
    """Open a span per consumed message, linked to the publish.

    Args:
        mq (MessageBroker): The broker, before connect().
    """
    mq.enable_tracing()

publish() already injects the traceparent and the current request id into the headers; enable_tracing() is the other half.

Link, not child

The consumer's span references the publish as a link, not as a parent. The semantic conventions recommend that for asynchronous consumption, and the reason is practical: the consumer may run minutes later, and a child span of that duration would stretch the request's trace and make its latency unreadable.

The request id is worth more than the span day to day

RequestIDMiddleware already puts the id on every HTTP log line. The worker now adopts the publisher's id while processing, so grep alone correlates request and consumption — without opening Jaeger.

Without the [otel] extra all of this is a no-op; request-id propagation works either way, because it does not depend on OpenTelemetry.

Background tasks — `TaskQueue`¶

A task queue takes slow work out of the request and hands it to a worker. TaskIQ does this but spreads the API across a broker, a scheduler, a schedule source and .kiq(). TaskQueue folds it all into one object with an obvious vocabulary.

Install with [tasks] (pulls taskiq + taskiq-aio-pika).

# src/tasks/__init__.py

from tempest_fastapi_sdk import EmailUtils
from tempest_fastapi_sdk.tasks import TaskQueue

from src.core.settings import settings

email = EmailUtils(**settings.email_kwargs())


tq = TaskQueue.rabbitmq(settings.TASKIQ_BROKER_URL)


@tq.task
async def send_welcome(to: str, name: str) -> None:
    """Runs on a worker, off the request path."""
    await email.send(to, "Welcome!", f"Hi, {name}.")

email is your mailer

email here is your e-mail sender — an EmailUtils instance (the [email] extra) wired up at module level. See the email recipe; swap it for your own send dependency.

@tq.task returns a typed Task object with two clear actions:

import asyncio

from src.db.models import UserModel
from src.tasks import send_welcome

email = "ana@example.com"
user = UserModel(name="Ana", email=email)


async def main() -> None:
    """Run this example."""
    # Enqueue to the worker and return immediately (the HTTP response doesn't wait):
    await send_welcome.enqueue(to=user.email, name=user.name)

    # Run inline, right here, returning the real value (handy in tests / reuse):
    await send_welcome.run(to="a@b.com", name="Ana")


asyncio.run(main())

enqueue instead of .kiq

enqueue() makes it obvious what happens: the call goes to the worker. run() executes the body locally, no broker. The cryptic .kiq name stays hidden (still reachable at send_welcome.taskiq_task if you need it).

Lifespan mirrors the message broker:

# src/api/app.py

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from src.tasks import tq


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    await tq.connect()
    try:
        yield
    finally:
        await tq.disconnect()

Tests without a broker

TaskQueue.memory() uses TaskIQ's in-memory broker: enqueue() runs the task immediately, in-process. No worker, no connection. run() always works, even without connect().

Worker resources — `on_startup` / `on_shutdown`¶

FastAPI's lifespan does not run in the worker. Without it the taskiq worker process has nowhere to open the database, the message broker or an HTTP client — and nowhere to close them: it works by accident, on the lazy connect of the first query, and never disposes the pool on the way out.

The hooks are that place:

# src/tasks/__init__.py

from tempest_fastapi_sdk.tasks import TaskQueue

from src.api.dependencies.resources import db
from src.core.settings import settings

tq = TaskQueue.rabbitmq(settings.TASKIQ_BROKER_URL)


@tq.on_startup
async def _open_resources() -> None:
    """Open the database when the worker starts."""
    await db.connect()


@tq.on_shutdown
async def _close_resources() -> None:
    """Dispose the pool when the worker stops."""
    await db.disconnect()

For the common case — resources that already speak connect / disconnect — the same thing fits on one line:

# src/tasks/__init__.py

from tempest_fastapi_sdk.tasks import TaskQueue

from src.api.dependencies.resources import broker, db
from src.core.settings import settings

tq = TaskQueue.rabbitmq(settings.TASKIQ_BROKER_URL, resources=[db, broker])

AsyncDatabaseManager, MessageBroker and AsyncMinIOClient satisfy the LifecycleResource protocol; any object of yours with both methods does too. They are opened left to right and closed right to left, so one that depends on an earlier one still finds it alive while closing. After construction, tq.use(db) does the same.

Scope: the worker by default

The hook is registered on TaskIQ's WORKER_STARTUP, so it does not fire in the web process — which has its own lifespan. Pass scope="client" (or "both") when you want the opposite: @tq.on_startup(scope="both").

Testable without a worker

The in-memory broker runs both sides' events in one process, so TaskQueue.memory(resources=[db]) executes the worker hooks on connect() / disconnect(). That is how this section's test checks the open and close ordering.

Class-based tasks¶

Symmetric to consumers: group tasks in a class with TaskDef. tq.register(...) returns a Task (constructor form) or a dict of Task keyed by method (grouped form).

import asyncio

from tempest_fastapi_sdk.tasks import TaskDef, task_method

from src.tasks import tq


# Constructor form — one task; name in the constructor, override run:
class NightlyReport(TaskDef):
    def __init__(self) -> None:
        super().__init__(name="reports:nightly")

    async def run(self, day: str) -> None:
        ...


nightly = tq.register(NightlyReport())        # -> Task


async def main() -> None:
    """Run this example."""
    await nightly.enqueue(day="2026-07-05")


    # Grouped form — many tasks, each method marked with @task_method:
    class ReportTasks(TaskDef):
        @task_method(name="reports:nightly")
        async def nightly(self, day: str) -> None: ...

        @task_method()
        async def weekly(self) -> None: ...


    tasks = tq.register(ReportTasks())            # -> {"nightly": Task, "weekly": Task}
    await tasks["nightly"].enqueue(day="2026-07-05")


asyncio.run(main())

The @tq.task function decorator is still available — both styles coexist.

Periodic tasks — `cron` / `interval`¶

Scheduling is part of the same TaskQueue — no separate scheduler in your code.

Don't know cron? Use the enums and helpers (v0.94.0)

Nobody should hand-write "0 9 * * MON-FRI". The tempest_fastapi_sdk.tasks module ships Cron (ready-made expressions), CronOffset (timezones by place, not digits), Weekday and builder functions (daily, weekdays, hourly, every_n_minutes, weekly, weekends, monthly). Each returns a plain cron string that drops straight into @tq.cron(...).

# src/tasks/__init__.py

from tempest_fastapi_sdk.tasks import Cron, CronOffset, Weekday, daily, weekdays

from src.tasks import tq


# Readable, no cron syntax:
@tq.cron(Cron.EVERY_WEEKDAY_9AM, cron_offset=CronOffset.BRASILIA)
async def daily_digest() -> None:
    ...


@tq.cron(daily(hour=9), cron_offset=CronOffset.BRASILIA)   # 09:00 BRT
async def other_digest() -> None:
    ...


@tq.cron(weekdays(hour=8, minute=30), cron_offset=CronOffset.BRASILIA)
async def morning_sync() -> None:
    ...


@tq.cron(Cron.EVERY_5_MINUTES)
async def heartbeat() -> None:
    ...

To run…	Write
Every 5 min	`Cron.EVERY_5_MINUTES` or `every_n_minutes(5)`
Daily at 9am	`daily(hour=9)`
Weekdays at 8:30	`weekdays(hour=8, minute=30)`
Every Monday	`weekly(Weekday.MON)`
First of the month	`monthly(day=1)`
In Brasília time	`cron_offset=CronOffset.BRASILIA`

CronOffset covers Brazil's timezones by name — BRASILIA (-03:00), FERNANDO_DE_NORONHA (-02:00), MANAUS (-04:00), ACRE (-05:00) — plus UTC. Prefer raw cron or intervals? Still supported:

from datetime import timedelta

from src.tasks import tq


@tq.cron("*/5 * * * *")                        # raw cron string
async def raw_cron() -> None:
    ...


@tq.interval(seconds=30)                        # every 30s
async def poll_remote() -> None:
    ...


@tq.interval(timedelta(minutes=15))
async def warm_cache() -> None:
    ...

In dev / single-process, run the scheduler inside the app:

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from src.tasks import tq


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    await tq.connect()
    await tq.start_scheduler()     # dev / single-process
    try:
        yield
    finally:
        await tq.stop_scheduler()
        await tq.disconnect()

The scheduler only enqueues — it doesn't execute

cron/interval enqueue the task into the same broker; a worker must be running to consume it. With no worker, triggers pile up in the queue.

Production: exactly one scheduler

start_scheduler() runs inside the FastAPI process — fine for dev. With multiple workers, each replica would run its own scheduler and duplicate every trigger. In production run one standalone scheduler and the workers separately.

Task reliability and observability¶

A worker-only task needs three things TaskIQ has but scatters: retry, dead-letter, and metrics. TaskQueue exposes all three as opt-in middleware — call them before connect(), and nothing touches the broker's middleware API.

Typed retry¶

RetryPolicy carries the retry config as labels; enable_retries() installs the TaskIQ middleware that reads them:

from tempest_fastapi_sdk.tasks import RetryPolicy, TaskQueue

tq: TaskQueue = TaskQueue.rabbitmq("amqp://guest:guest@localhost:5672/")
tq.enable_retries(default_max_retries=3)


@tq.task(name="reports:nightly", retry=RetryPolicy(max_retries=5))
async def nightly() -> None:
    ...   # re-run up to 5x on error

Dead-letter — where terminal failures go¶

When a task fails with no retry configured, or after its retries are exhausted, the call goes to your DeadLetterSink exactly once. The target is yours — a MessageBroker channel, a DB row, an alert. The SDK assumes no backend:

from tempest_fastapi_sdk.tasks import DeadLetter, TaskQueue

from src.queue import mq


tq: TaskQueue = TaskQueue.rabbitmq("amqp://guest:guest@localhost:5672/")


async def to_dlq(dead_letter: DeadLetter) -> None:
    await mq.publish("tasks.dead", {
        "task": dead_letter.task_name,
        "args": dead_letter.args,
        "error": str(dead_letter.exception),
        "retries": dead_letter.retries,
    })


tq.dead_letter(to_dlq, default_max_retries=3)

Pair it with retry

Pass the same default_max_retries to enable_retries and dead_letter so the "retries exhausted" point lines up for tasks that set no explicit max_retries. Install order does not matter — the dead-letter middleware decides on its own by reading the message labels.

Per-task Prometheus metrics¶

TaskMetrics counts executions (by status) and a duration histogram, labelled by task, into the same /metrics the SDK already serves (pass the shared registry):

from tempest_fastapi_sdk.tasks import TaskMetrics, TaskQueue

tq: TaskQueue = TaskQueue.rabbitmq("amqp://guest:guest@localhost:5672/")
tq.enable_metrics(TaskMetrics())   # tasks_runs_total{task,status} + tasks_duration_seconds{task}

Dead-letter panel in the admin¶

The DeadLetterSink says what to do with a failure; to see and re-run failures, persist them to a table and surface them in the admin. DbDeadLetterSink writes each terminal failure; make_dead_letter_admin_model builds a read-mostly AdminModel (filter by task, search the error, export) with an optional requeue bulk action.

from tempest_fastapi_sdk.admin import AdminSite
from tempest_fastapi_sdk.tasks import (
    DbDeadLetterSink,
    TaskQueue,
    make_dead_letter_admin_model,
    make_dead_letter_model,
)

from src.core.resources import db   # AsyncDatabaseManager


DeadLetterModel = make_dead_letter_model()   # or subclass BaseDeadLetterModel by hand

tq: TaskQueue = TaskQueue.rabbitmq("amqp://guest:guest@localhost:5672/")
tq.dead_letter(DbDeadLetterSink(db, DeadLetterModel))   # persist terminal failures

site: AdminSite = AdminSite(title="Ops")
site.register(make_dead_letter_admin_model(DeadLetterModel, tq=tq))   # panel + requeue

Passing tq= wires the requeue action: the operator selects rows, each call is re-enqueued with its stored args / kwargs, and the requeued rows are deleted.

No Flower clone

TaskIQ exposes no live queue state (Flower is Celery-specific), so this panel does not try to show pending/in-flight jobs. It shows what is real and persisted: the terminal failures.

For a "what tasks exist" inventory, task_inventory(tq) returns list[TaskInfo] (name / schedule / retry) read straight off the broker — serve it as JSON, a log, or your own page:

from tempest_fastapi_sdk.tasks import task_inventory

from src.tasks import tq


for info in task_inventory(tq):
    print(info.name, info.schedule, info.retry_on_error, info.max_retries)

Workers in production¶

The worker and the scheduler are separate processes pointing at the raw objects TaskQueue exposes. The TaskIQ CLI resolves module:attribute with a plain getattr, so bind both to module-level names — a dotted path does not resolve:

# src/tasks.py — after registering the tasks
from tempest_fastapi_sdk.tasks import TaskQueue

tq: TaskQueue = TaskQueue.rabbitmq("amqp://guest:guest@localhost:5672/")

broker = tq.broker
scheduler = tq.scheduler

# consumes and executes the tasks
taskiq worker    src.tasks:broker

# a single scheduler process for the whole cluster
taskiq scheduler src.tasks:scheduler

tq.broker is the TaskIQ broker (it knows every registered task); tq.scheduler is the internal TaskiqScheduler.

src.tasks:tq.broker does not work

The CLI runs getattr(module, "tq.broker") and raises AttributeError: module 'src.tasks' has no attribute 'tq.broker' — the process never starts. The same holds for any dotted path (tq.scheduler, scheduler.scheduler). Hence the broker = tq.broker above.

Transactional outbox¶

When a handler writes a row AND publishes an event, doing them separately is unsafe: a crash between the commit and the publish loses the event; between the publish and the commit creates a phantom event. The outbox pattern writes the business row and an outbox row in the same transaction — either both commit or neither. A relay then reads the outbox and publishes to the broker later.

The SDK already ships the primitive

Unlike what the old version of this page said, the outbox is an SDK primitive: BaseOutboxModel (the table), OutboxRelay (the worker that drains and publishes, with exponential backoff and FOR UPDATE SKIP LOCKED on Postgres) and BaseRepository.save_with_outbox (the write side). The relay takes any async publish — it plugs straight into MessageBroker:

# src/tasks/__init__.py — outbox relay
from tempest_fastapi_sdk import OutboxRelay

from src.db.models import OutboxModel
from src.queue import mq          # MessageBroker
from src.core.resources import db  # AsyncDatabaseManager


relay = OutboxRelay(
    db,
    model=OutboxModel,
    # channel first, payload second — the same publish signature:
    publish=lambda event: mq.publish(event.topic, event.payload),
)

# In the lifespan (or as a dedicated process): drains until cancelled.
# asyncio.create_task(relay.run(poll_interval=1.0))

The full guide — model, producer service with save_with_outbox, retention and concurrency — lives in the dedicated Outbox recipe.

Recap / next steps¶

MessageBroker — typed, transport-agnostic pub/sub over FastStream: @mq.on("channel") + await mq.publish("channel", model). At-least-once fan-out across services.
TaskQueue — tasks over TaskIQ: @tq.task → await task.enqueue(...) (to the worker) or await task.run(...) (inline). .memory() for tests.
@tq.cron / @tq.interval — periodic on the same object; start_scheduler() in dev, standalone CLI in production.
Cron without syntax — Cron / CronOffset / Weekday + helpers (daily, weekdays, every_n_minutes, …) to schedule by name; CronOffset.BRASILIA instead of "-03:00".
Styles — decorators (@mq.on, @tq.task, @tq.cron) or classes (Consumer + mq.register, TaskDef + tq.register); both coexist.
Outbox — BaseOutboxModel + OutboxRelay + save_with_outbox, with the relay's publish pointing at MessageBroker. See Outbox.
Rename (v0.94.0) — AsyncBrokerManager → AsyncQueueManager (thin wrapper; old alias kept). The MessageBroker / TaskQueue facades stay recommended; AsyncTaskBrokerManager / AsyncTaskScheduler remain functional legacy.