Stop Fighting Celery: Async Task Queues in FastAPI with Taskiq

[embed]Hapkiduki/taskiq-demo demo about the article Stop Fighting Celery: Async Task Queues in FastAPI with Taskiq - Hapkiduki/taskiq-demogithub.com

Your FastAPI codebase is async def all the way down. asyncpg, httpx, everything awaits beautifully. Then you need background jobs, and you reach for Celery — because that's what everyone reaches for.

And suddenly you’re maintaining two programming models in one codebase.

Before the objection: yes, FastAPI ships BackgroundTasks, and its docs use exactly the send-an-email example. It's the right tool right up until losing a job stops being acceptable — it runs inside your API process, so a restart mid-send silently drops the work, with no retries, no persistence, and no way to scale it apart from your web traffic. The moment a job must survive a deploy, retry with backoff, or run on a schedule, you've outgrown it.

That’s where Taskiq comes in: a task queue built async-first. What follows is a compact but production-shaped setup — routed queues, retries with backoff, scheduling, dependency injection, and a test suite that needs no infrastructure at all. Every snippet is from a runnable repo.

The uncomfortable part about Celery and async

Credit where it’s due: Celery has fifteen-plus years in production, an enormous ecosystem, and active maintenance — 5.6 shipped in early 2026. On synchronous Django or Flask it’s still a fine answer, and nothing in Taskiq matches Canvas for complex workflow graphs.

But Celery’s worker is synchronous. Native async def support has been requested since 2017 and still isn't there. The workarounds all hurt: third-party worker pools, from-scratch reimplementations, or calling asyncio.run() inside every task — a fresh event loop per invocation, and your async service layer treated like a foreign country.

That’s the real cost. Not performance — friction. Two other things Taskiq gets right, which I’d argue matter as much:

• Typed kicks. send_welcome.kiq(email=...) mirrors the task's real signature, so your IDE completes it and mypy checks it. Celery's .delay() is Any all the way down, which makes renaming a task argument a grep-and-pray operation.
• An honest test story. Celery’s task_always_eager runs tasks through a code path production never uses, skipping the broker and serialization entirely. Taskiq lets you swap in a real broker with an in-memory transport — same middlewares, same message format, executed inline.

One broker, a queue per workload

The design rule: queues follow workloads, not function names. Emails are light and latency-sensitive; report crunching is heavy and can lag. Separate queues let you scale, prioritize, and drain them independently.

The broker is a factory that branches on environment:

def build_broker() -> AsyncBroker:
    middlewares = [
        _retry_middleware(schedule_source),
        AttemptMiddleware(),
        TimingMiddleware(),
    ]
    if os.environ.get("APP_ENV") == "test":
        # await_inplace: kicked tasks run synchronously inside the caller's
        # await, so tests observe side effects deterministically.
        return InMemoryBroker(await_inplace=True).with_middlewares(*middlewares)
    # Imported lazily: the test path above must work without the AMQP/Redis
    # stack installed or reachable.
    from taskiq_aio_pika import AioPikaBroker
    ...

Two deliberate moves worth naming. First, factory plus module-level singleton: every process — API, workers, scheduler, tests — imports the same broker object, so transport decisions live in exactly one place. Second, lazy imports. PEP 8 says imports go on top, and that's the right default; the valid exception is an import that's expensive, optional, or environment-dependent. This module runs its factory at import time, so a top-level taskiq_aio_pika would force the whole AMQP stack onto a CI runner that only wants to exercise the in-memory path.

The production branch declares the real topology — one topic exchange, a durable queue per workload, a shared dead-letter queue, and a Redis result backend with a TTL. And the queue list comes from an env var:

raw = os.environ.get("WORKER_QUEUES", "")
queue_names = [q.strip() for q in raw.split(",") if q.strip()] or list(ALL_QUEUES)

That’s my favourite line in the project. Every worker runs the same command; the only difference between “the emails worker” and “the heavy worker” is WORKER_QUEUES in its environment. Worker topology lives in your deployment manifest, so adding a dedicated worker for a hot queue is an infra change, not a code change.

Tasks wear their behaviour as labels

Extra keyword arguments on the decorator become labels — metadata that travels with every message. Routing, retries, and cron are all labels:

@broker.task(task_name="emails.send_welcome", queue_name=QUEUE_EMAILS)
async def send_welcome(email: str, ...) -> None:
    ...
@broker.task(
    task_name="reports.daily_summary",
    queue_name=QUEUE_PERIODIC,
    schedule=[{"cron": "0 8 * * *"}],  # every day at 08:00 UTC
)
async def daily_summary(...) -> dict[str, int]:
    ...

The cron sits on the task: no central beat file collecting merge conflicts, and deleting the task deletes its schedule, so zombie crons are structurally impossible. (For schedules that can’t exist at deploy time — “remind this user in ten minutes” — Taskiq has Redis-backed dynamic sources with schedule_by_time, schedule_by_cron, and a handle you can unschedule().)

Set task_name explicitly, by the way. Otherwise it's derived from the module path, and the day you move that file, messages already sitting in queues reference a name that no longer exists.

Retries, and knowing when to stop

Here’s the pattern most retry tutorials skip — how to stop gracefully:

attempt = int(context.message.labels.get("attempt", 1))
    ...
    except ConnectionError:
        if attempt >= MAX_DELIVERY_ATTEMPTS:
            FAILED_WEBHOOKS.append(delivery_id)
            return "failed"
        raise

It fits in a sentence: raise to retry, return to stop. An exception tells the retry middleware to re-enqueue. But on the final attempt we catch our own error, record the permanent failure where the product can see it, and return. The message gets acknowledged; the loop ends on our terms instead of dumping the failure into a result backend nobody watches.

Use SmartRetryMiddleware, not the SimpleRetryMiddleware most examples reach for. Simple retries fire immediately and in lockstep, so a downstream outage means every failed task hammers the recovering service at the same instant. Smart retries wait, back off exponentially, and add jitter to spread the herd.

One catch I only found by reading the broker source. Backoff needs somewhere to hold a task between attempts, and RabbitMQ’s delay_queue looks like the obvious answer — but taskiq-aio-pika declares it with a single fixed dead-letter routing key, the first task queue. I verified it: a delayed heavy task came back into emails. With more than one workload queue, every delayed retry gets misrouted. So pass a schedule_source instead — the retry is stored in Redis and re-kicked by the scheduler with its original labels intact:

return SmartRetryMiddleware(
        default_retry_count=3,
        default_delay=2,
        use_jitter=True,
        use_delay_exponent=True,
        max_delay_exponent=60,
        schedule_source=schedule_source,
    )

The trade: retries now depend on the scheduler process. When it’s down they wait in Redis rather than vanish, which I’ll take.

Custom middlewares, and a lesson about underscores

The middleware ABC is small enough to memorize: pre_send/post_send on the producer, pre_execute/post_execute/on_error on the worker. Cross-cutting concerns — timing, metrics, tracing, log context — live here instead of being pasted into every task.

Which raises something a sharp reviewer would flag. The retry count lives in a _retries label, and that underscore means what it means everywhere in Python: internal, not yours. Taskiq exposes no public API for the attempt number — even its own OpenTelemetry middleware reads _retries directly. So should your business logic?

No. A middleware reading its own private label is fine; a task reading another component’s private label couples your business logic to a library’s internals, and the day it’s renamed your attempt count silently becomes “always 1”. When there’s no public API, contain the trespass in one place:

class AttemptMiddleware(TaskiqMiddleware):
    def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
        message.labels["attempt"] = int(message.labels.get("_retries", 0)) + 1
        return message

Six lines. Tasks depend only on the public attempt label, the retry tests pin the semantics, and a future rename breaks loudly in CI instead of quietly in production.

Dependency injection on both sides of the queue

FastAPI’s Depends is half the reason people love the framework. Taskiq ships its twin, TaskiqDepends, with the same model: provider functions, resolved per execution, cached within it, generator setup/teardown included.

The move that makes it pleasant is writing providers that know about neither framework:

def get_email_client() -> EmailClient:
    """One client per process: the API and each worker build their own."""
    return _email_client

The handler injects it the FastAPI way, Annotated[EmailClient, Depends(get_email_client)]; the task injects the same provider the Taskiq way, client: EmailClient = TaskiqDepends(get_email_client). One resource definition, consumed from HTTP handlers and background tasks alike.

That asymmetry is deliberate, by the way. .kiq() mirrors the task's signature for the type checker, so an injected parameter needs a default — otherwise every call site gets asked for an EmailClient it shouldn't have to know about.

The test suite that needs nothing

This is where the factory’s test branch pays off. The conftest sets one env var before any app import:

# Must run BEFORE any `app` import: app/broker.py builds the broker at import
# time, and APP_ENV=test is what selects InMemoryBroker(await_inplace=True).
os.environ["APP_ENV"] = "test"

With await_inplace=True, every .kiq() runs the task to completion inside the caller's await — through the real middlewares, with the real message format. No sleeps, no polling, no flakes:

async def test_retry_until_success() -> None:
    await deliver_webhook.kiq(delivery_id="d-1", fail_times=1)

    # First attempt raises, SmartRetryMiddleware re-kicks, second succeeds.
    assert WEBHOOK_ATTEMPTS["d-1"] == 2
    assert FAILED_WEBHOOKS == []

async def test_terminal_failure_stops_retrying() -> None:
    await deliver_webhook.kiq(delivery_id="d-2", fail_times=99)

    assert WEBHOOK_ATTEMPTS["d-2"] == MAX_DELIVERY_ATTEMPTS
    assert FAILED_WEBHOOKS == ["d-2"]

Sit with the first one. It asserts the task executed exactly twice: the retry middleware really ran, the re-kicked message really carried its incremented counter, the second attempt really succeeded. That’s not a mock of the retry system — it is the retry system, minus RabbitMQ. The second test pins max_retries semantics (three total executions, not one plus three) so a future version change breaks CI instead of production.

The suite also covers the full round trip: POST returns 202 with a task id, the task runs inline, GET reads the computed result from the in-memory backend. Seven tests, under half a second, no Docker installed. The honest answer to “how do I test my Celery tasks?” usually involves a docker-compose file; here it’s pytest.

Seeing it with your own eyes

My favorite demo for a skeptical teammate. Start only the heavy-queue worker, then POST /signups. The API cheerfully returns 202 — and the welcome-email message lands in the emails queue and waits. The management UI shows emails: 1 message ready. Nothing crashed, nothing was lost, and reports still process because their worker is alive.

Start an emails worker and it drains instantly. That parked message is the whole pitch for queue-based architecture, holding still long enough to look at: failure isolation, and work waiting for workers instead of the other way around.

Production notes worth keeping

• Prefetch (qos) is a per-consumer buffer, not a concurrency setting. Generous for cheap tasks; small — even 1 — for long ones, or a busy worker hoards messages an idle one could take.
• Watch the dead-letter queue. Depth should be zero; alert the moment it isn’t. Cheapest, highest-signal metric in the system.
• Run exactly one scheduler. No leader election, so two replicas fire every cron twice.
• Make tasks idempotent. The default ack type is when_saved, so a worker that dies mid-task gets its message redelivered.
• The producer must start the broker. The worker CLI calls broker.startup() for you; your API process does it in the lifespan, guarded with if not broker.is_worker_process:.

So, should you switch?

On sync Django, or with workflow graphs that look like org charts — stay on Celery, sincerely.

But if you’re building on FastAPI, your codebase already made its concurrency decision, and every day your task queue disagrees you pay a small tax: bridge code, untyped kicks, eager-mode tests you don’t quite trust. Taskiq removes it. Coroutine tasks that call your async services natively, labels keeping routing and retries and cron next to the code, DI that mirrors FastAPI’s, and a test story that’s honestly better than what most of us had with Celery.

About 200 lines of application code. No glue, no second programming model.

[embed]Hapkiduki/taskiq-demo demo about the article Stop Fighting Celery: Async Task Queues in FastAPI with Taskiq - Hapkiduki/taskiq-demogithub.com