Celery vs Dramatiq vs arq: Picking a Task Queue for FastAPI

Why FastAPI BackgroundTasks is not a task queue

A task queue decouples requesting work from doing it. The web process serializes a job description into a durable broker such as Redis or RabbitMQ and returns immediately. Separate worker processes pull jobs, execute them, and acknowledge completion. Because the job lives outside the web process, it can be retried after a failure, delayed, rate-limited, and scaled on its own hardware. That is the reason to use one: sending email, generating reports, calling slow third-party APIs, or anything that takes longer than a request should block or must not be lost.

FastAPI's BackgroundTasks does something much narrower. It is inherited from Starlette and runs a function after the response is sent, inside the same process that served the request. Async functions run on the server's event loop, and sync functions run in its threadpool. Nothing is written to storage. If the function raises, there is no retry. If the pod is killed by a deploy, an OOM, or a scale-down before the function finishes, the work is lost and no record of it remains. CPU-heavy work also competes directly with request handling. BackgroundTasks is a good fit for short, best-effort side effects you can afford to drop, such as non-critical logging or cache warming.

When work has to survive restarts, be retried, or scale apart from the API, you need an external broker and dedicated workers. This article compares three options: Celery, Dramatiq, and arq. The argument is that you should not choose by comparing feature lists, because all three cover the common cases. Choose by two things: whether your task code is async or blocking, and how much operational surface your team is prepared to run.

How each library works: sync workers vs an asyncio event loop

All three share the same outer shape. A FastAPI endpoint serializes a message describing the job and pushes it to a broker. A separate worker process pulls messages off and runs your function. The difference that matters is what "runs your function" means inside that worker.

Celery and Dramatiq: synchronous callables

Celery's default pool forks child processes, and each child executes one task at a time. Dramatiq runs several processes, each with a pool of threads. In both, a task is an ordinary blocking function. That fits code built on synchronous drivers and requests. It works against a FastAPI codebase whose database layer, HTTP clients and SDKs are all async.

The usual workaround is to call asyncio.run() inside each task. That creates and closes a fresh event loop on every invocation, which causes two problems:

• Pools stop working. Connection pools and clients such as an async SQLAlchemy engine or an httpx.AsyncClient are bound to the loop they were created on. A module-level pool built under one loop fails, or quietly misbehaves, when a later task uses it under a new loop.
• Connections churn. Recreating the pool per task means opening new connections on every job and leaking any you forget to close.

The more robust fix is one long-lived loop per worker, run in a background thread, with tasks submitting coroutines to it. That works, but you now own its startup, shutdown and error propagation.

arq: coroutines on one loop

An arq worker runs a single asyncio event loop and executes tasks as coroutines, with up to max_jobs running at once. Startup and shutdown hooks let you build pools and clients once and share them across every job. As a result, the same async code your endpoints use runs unchanged in the worker.

Copy     +------------------+   enqueue    +----------+
     | FastAPI endpoint |------------->|  Broker  |
     +------------------+  (message)   +----------+
                                            |
               +----------------------------+------+
               |                                   |
               v                                   v
+-----------------------------+     +-----------------------------+
| Celery / Dramatiq worker    |     | arq worker                  |
| processes or thread pool    |     | one asyncio event loop      |
|                             |     |                             |
| task(): plain sync function |     | startup hook: build pools   |
|   asyncio.run(coro) per job |     | job1  job2  ... <= max_jobs |
|   -> new loop on every call |     |   |     |                   |
|   -> pool tied to dead loop |     |   v     v                   |
|   -> connections churn/leak |     | shared async DB/HTTP pools  |
|                             |     | shutdown hook: close pools  |
+-----------------------------+     +-----------------------------+

FastAPI sends jobs through a broker to two kinds of worker: Celery and Dramatiq run sync task functions that need asyncio.run() to call async code, while arq runs up to max_jobs coroutines on one event loop that shares pools created once at startup.

Implementing tasks and enqueuing from async FastAPI endpoints

In arq, a task is an async def whose first argument is ctx, the dictionary populated by your worker's startup hook. You enqueue by name with await redis.enqueue_job("send_invoice", invoice_id), where redis is the pool returned by arq.create_pool. Create that pool once in FastAPI's lifespan handler, store it on app.state, and close it on shutdown. Opening a connection per request adds avoidable latency and churns Redis connections under load.

Celery and Dramatiq are both declared with decorators: @app.task for Celery and @dramatiq.actor for Dramatiq. You enqueue with task.delay(...), or apply_async(...) when you need options such as countdown or a specific queue, and with actor.send(...) for Dramatiq. All three calls do blocking network I/O against the broker. Usually that is a sub-millisecond write to a nearby Redis or RabbitMQ and nobody notices. When the broker is slow, failing over, or applying backpressure, though, the call stalls the event loop, and every request on that worker stalls with it. Wrapping the call in starlette.concurrency.run_in_threadpool or anyio.to_thread.run_sync limits the damage to one thread. The cost is a thread handoff per enqueue.

Whichever library you pick, route every enqueue through a small interface of your own, for example an async def enqueue(task_name, **kwargs) on a JobQueue class. Endpoints then depend on that interface and never import Celery, Dramatiq or arq directly. The threadpool decision lives in one place. Tests can swap in an in-memory fake. A later migration becomes a change to one adapter instead of edits to every call site.

Copyfrom contextlib import asynccontextmanager
from typing import Any, Protocol

from arq import create_pool
from arq.connections import ArqRedis, RedisSettings
from celery import Celery
from fastapi import FastAPI, Request
from starlette.concurrency import run_in_threadpool

class JobQueue(Protocol):  # endpoints depend only on this
    async def enqueue(self, task_name: str, **kwargs: Any) -> None: ...

class ArqQueue:
    def __init__(self, pool: ArqRedis) -> None:
        self.pool = pool
    async def enqueue(self, task_name: str, **kwargs: Any) -> None:
        await self.pool.enqueue_job(task_name, **kwargs)  # native async I/O

class CeleryQueue:
    def __init__(self, app: Celery) -> None:
        self.app = app
    async def enqueue(self, task_name: str, **kwargs: Any) -> None:
        # send_task does blocking broker I/O; keep it off the event loop
        await run_in_threadpool(self.app.send_task, task_name, kwargs=kwargs)

@asynccontextmanager
async def lifespan(app: FastAPI):
    pool = await create_pool(RedisSettings(host="redis.internal", port=6379))
    app.state.jobs = ArqQueue(pool)  # one pool per process, not per request
    # Migration path: app.state.jobs = CeleryQueue(Celery(broker="redis://redis.internal:6379/0"))
    yield
    await pool.close()

app = FastAPI(lifespan=lifespan)

@app.post("/invoices/{invoice_id}/send")
async def send_invoice(invoice_id: int, request: Request):
    jobs: JobQueue = request.app.state.jobs
    await jobs.enqueue("send_invoice", invoice_id=invoice_id)
    return {"status": "queued"}

A thin JobQueue interface with an arq adapter, whose pool is created once in the FastAPI lifespan handler, and a Celery adapter that moves the blocking send_task call into a threadpool. Switching libraries changes only which adapter the lifespan handler installs.

Brokers, delivery semantics and failure handling

Behind that adapter, the broker determines what happens when a worker dies mid-task. All three libraries give you at-least-once delivery at best, so assume every task can run twice.

Celery's default early ack means a crashed worker loses the task. With acks_late=True the task is redelivered instead, which is usually what you want. On Redis, "redelivered" means after the visibility timeout expires. Set it longer than your slowest task, or Celery will hand a still-running task to a second worker.

Redis also needs persistence (AOF), maxmemory-policy noeviction so queued jobs are never evicted, and an accepted risk that async replication can drop recent writes on failover.

A sane baseline for each library:

• Celery: acks_late, bounded autoretry_for with backoff, and explicit soft and hard time limits.
• Dramatiq: reduce the default retry count and set max_retries=0 for anything non-idempotent. Otherwise a timeout or transient error quietly re-runs it.
• arq: a tight job timeout and an explicit max-tries value.

In every library, key side effects on an idempotency token, such as a unique constraint or a "processed" record checked before acting.

Copy# --- Celery: late ack, bounded retries, explicit time limits ---
from celery import Celery

celery_app = Celery("app", broker="redis://redis.internal:6379/0")
# Redis redelivers unacked tasks after this many seconds; keep it above the slowest task
celery_app.conf.broker_transport_options = {"visibility_timeout": 3600}

@celery_app.task(acks_late=True, autoretry_for=(ConnectionError,), retry_backoff=True,
                 max_retries=5, soft_time_limit=60, time_limit=90)
def send_receipt(order_id: str) -> None:
    ...  # skip if a receipt row with this order_id already exists

# --- Dramatiq: retries and time limits are on by default, so set them deliberately ---
import dramatiq
from dramatiq.brokers.redis import RedisBroker

dramatiq.set_broker(RedisBroker(url="redis://redis.internal:6379/1"))

@dramatiq.actor(max_retries=3, min_backoff=1_000, time_limit=60_000)  # values in ms
def sync_inventory(sku: str) -> None:
    ...  # idempotent: safe to re-run

@dramatiq.actor(max_retries=0, time_limit=30_000)  # not idempotent: never auto re-run
def charge_card(payment_id: str) -> None:
    ...  # insert payment_id under a unique constraint before calling the provider

# --- arq: explicit Retry, tight job timeout, capped tries ---
from arq import Retry

async def refresh_cache(ctx, key: str) -> None:
    try:
        ...
    except ConnectionError:
        raise Retry(defer=ctx["job_try"] * 5)  # back off 5s, 10s, ... per attempt

class WorkerSettings:
    functions = [refresh_cache]
    job_timeout = 60  # seconds
    max_tries = 3

Side-by-side baseline policies: Celery with acks_late, bounded autoretry and soft/hard limits plus a Redis visibility timeout; Dramatiq with tightened retries and max_retries=0 for a non-idempotent actor; arq with a Retry exception, job timeout and max_tries.

Scheduling, workflows and concurrency sizing

Periodic jobs

Celery ships beat, a separate scheduler process. Run exactly one, or each schedule fires once per beat instance. arq declares cron jobs in the worker settings, and the Redis-backed job ID deduplicates runs across workers. Dramatiq has no scheduler, so you pair it with system cron or a small scheduler process that only enqueues messages.

Workflows

Celery has the richest primitives: chain, group, and chord. A chord needs a result backend, and chords are a common source of stuck or lost callbacks. Dramatiq offers pipelines and groups, and group completion callbacks need its results middleware. arq has none, so you enqueue the next step from inside a job. Before choosing a library for its primitives, check whether you really need fan-out/fan-in. Many "workflows" are a single job followed by a database status update.

Concurrency models

• Celery: prefork processes by default, with thread or green-thread pools as alternatives.
• Dramatiq: processes × threads.
• arq: one asyncio event loop per worker.

An arq worker runs everything on that loop, so a blocking call such as synchronous HTTP, time.sleep, or image resizing freezes every in-flight job. Offload that work to a process pool executor, or route it to a separate queue served by process-based workers.

Sizing

Sizing follows from how long jobs wait versus how long they compute.

Illustrative estimate, with assumed inputs of 50 jobs/s, 200 ms of I/O wait, and 5 ms of CPU per job:

• In-flight jobs: 50 × 0.2 = 10
• CPU needed: 50 × 0.005 = 0.25 cores

One arq worker allowing about 20 concurrent jobs covers this load.

Change the assumption to 100 ms of CPU per job and you need 5 cores. Threads and an event loop won't help because of the GIL, so plan for at least five worker processes.

Copyimport asyncio
import hashlib
from concurrent.futures import ProcessPoolExecutor

from arq import cron
from arq.connections import RedisSettings


def cpu_heavy(payload: bytes) -> str:
    # Stand-in for image resizing or parsing. Runs in a child process,
    # so it neither holds the GIL nor blocks the worker's event loop.
    for _ in range(200_000):
        payload = hashlib.sha256(payload).digest()
    return payload.hex()


async def process_upload(ctx, upload_id: str) -> str:
    await asyncio.sleep(0.2)  # I/O wait (HTTP, DB); other jobs run meanwhile
    loop = asyncio.get_running_loop()
    # Offload blocking CPU work instead of calling cpu_heavy() directly
    return await loop.run_in_executor(ctx["pool"], cpu_heavy, upload_id.encode())


async def nightly_cleanup(ctx) -> None:
    await asyncio.sleep(0.1)  # e.g. delete expired rows


async def startup(ctx) -> None:
    ctx["pool"] = ProcessPoolExecutor(max_workers=4)  # roughly one per core for CPU work


async def shutdown(ctx) -> None:
    ctx["pool"].shutdown(wait=True)


class WorkerSettings:
    functions = [process_upload]
    # Cron runs are deduplicated through Redis, so several workers can share this config
    cron_jobs = [cron(nightly_cleanup, hour=3, minute=0)]
    max_jobs = 20  # concurrent in-flight jobs on this worker's single event loop
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = RedisSettings(host="localhost", port=6379)

An arq worker that declares a cron job, caps concurrent jobs with max_jobs, and moves CPU-bound work to a ProcessPoolExecutor so it doesn't stall the shared event loop.

Operations, observability and the cost of ownership

Results belong in your database

All three libraries can store return values in Redis. Celery keeps them until they expire, and arq keeps them for a configurable window. That memory competes with your queues. On a busy service, store each outcome in the application database as part of the task, and disable result storage where you never read it. Celery's ignore_result does this per task.

Illustrative estimate (assumed numbers, not a benchmark): result memory ≈ jobs per hour × retention hours × bytes per result. With 200,000 jobs/hour, 24-hour retention and 2 KB per result, that is 200,000 × 24 × 2 KB ≈ 9.6 GB of Redis spent on data nobody queries.

Visibility and shutdown

Celery has Flower and the most third-party tooling. Dramatiq ships a Prometheus middleware that exports worker metrics. arq gives you logs and whatever you query from Redis yourself.

For deploys, the orchestrator's termination grace period must exceed your longest task. Otherwise SIGTERM turns into SIGKILL partway through a job. Test what each worker does with in-flight jobs on shutdown before you rely on it.

Configuration surface

Celery's defaults cost the most debugging time:

• Early acks. Messages are acknowledged before the task runs, so a crash loses the job.
• Prefetch. Each worker process reserves several messages, which starves idle workers when tasks are long.
• Redis visibility timeout. Unacknowledged or long-delayed tasks get redelivered and run twice.

Dramatiq acknowledges after processing and has fewer settings. arq has fewer still. The trade-off is arq's smaller community: when it breaks, you are more likely to be reading its source.

Decision matrix: which to use, when to avoid each, and alternatives

Work through these questions in order. The first one that produces a firm answer usually settles the choice.

Choose arq if your service is async-first, already depends on Redis, and its jobs mostly call databases or HTTP APIs. Avoid arq for CPU-heavy work, because one blocking task stalls every other job on that event loop. Also avoid it if your organisation has standardised on RabbitMQ.

Choose Dramatiq for synchronous workers that need to be reliable without much tuning. Avoid it if native async tasks matter more to you than its simpler model.

Choose Celery only when you will actually use its breadth: canvas workflows, multiple brokers, or a team that already knows its configuration well. Avoid it if you are a small team running simple jobs. You would pay its configuration and operational cost without using most of what it offers.

Keep using BackgroundTasks for work you can afford to lose, such as cache warming or non-critical notifications.

Plan for the cost of switching later. Task signatures, retry policies, result handling and monitoring all depend on the library you pick. Moving to a different one means rewriting the code that enqueues tasks and draining the jobs already in the queue.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to celery vs dramatiq vs arq, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.