Running ARQ Workers with FastAPI

Key takeaways:

• An ARQ task is an async function whose first positional argument is the worker's ctx dict, not one of your own arguments.
• The web process only ever does one thing: encode a job and write it to Redis through a pool opened in lifespan.
• An explicit _job_id makes enqueue_job return None instead of a second job, which is how double submissions collapse.
• Every argument crosses a process boundary as bytes, so serialization failures surface in the HTTP request, not in the worker.
• The worker is a separate process with its own pools, its own settings and its own scaling.

This guide implements the ARQ path described in Background Task Processing, and assumes you have already decided against in-process work using the comparison of BackgroundTasks, Celery and ARQ.

What Was Executed on This Page

Redis and arq are not installed in this site's verification environment, and there is no broker to talk to. Rather than invent worker output, the example below runs the half of the system that actually lives in your FastAPI process: the pool object on app.state, the enqueue_job call, deduplication by job id, and the exact encoded payload that would be handed to Redis. The store behind that interface is an in-process dict rather than a Redis connection, and the transcript says so. Everything attributed to the worker daemon — polling, retry backoff, cron scheduling, result expiry — is described in prose with no fabricated transcript.

That split is not a compromise. Almost every ARQ bug that reaches production is on the FastAPI side of the boundary: a job enqueued with an unserializable argument, a missing _job_id, a function name that does not match anything in WorkerSettings.functions. Those are exactly the parts that can be exercised here.

The Problem This Solves

You need deferred work that survives a deploy, retries on failure, and scales separately from the API — in a codebase that is async from top to bottom. Celery can do all of that, but it arrives with a synchronous execution model, a separate beat process for schedules, and a configuration surface that is large relative to the job. ARQ is Redis-only, async-native, and small enough to read in an afternoon.

The cost of that smallness is that ARQ assumes you understand the boundary it draws. Two processes, one shared Redis instance, and a strict rule that nothing but encodable data crosses between them.

Why It Works This Way

enqueue_job is not a function call that is deferred. It is a write. ARQ takes the function name as a string, packs it together with your arguments into a single encoded value, and stores that value in Redis under a key derived from the job id. The callable itself never moves. The worker resolves the name against the functions list in its own WorkerSettings at the moment it pops the job.

Three consequences follow directly, and they explain nearly every question people have about ARQ.

A typo in the function name is not caught at enqueue time by your type checker. You are passing a string. If the worker's functions list does not contain a matching name, the job is queued successfully and then fails on the worker with no clue at the call site. Keeping a module-level mapping of name to callable and enqueuing through it turns that into an error your web process raises immediately.

Arguments must be encodable, and the failure happens in your request. ARQ serializes by default with pickle, which is more permissive than JSON but still cannot handle an open connection, a SQLAlchemy session, or an object holding a lambda. Whatever the codec, the encode step runs inside enqueue_job — that is, inside your HTTP handler — so an unserializable argument is a 500 on the API, not a mysterious worker crash. That is the good outcome, and it is worth designing for by passing identifiers only.

Deduplication is a property of the key, not of the payload. ARQ writes the job under the id you supply. If that key already exists, it does not enqueue, and enqueue_job returns None rather than a Job. Two identical submissions therefore collapse — but only for as long as the key exists, which ends when the job finishes and its result expires. Deduplication buys you protection against a double-clicked button, not against a re-submission an hour later.

The Fix

Start with the task definitions and worker configuration. The ctx first argument is the piece that trips people up: it is supplied by the worker, so a task declared as async def sync_invoice(invoice_id: str) will receive the context dict as invoice_id and fail confusingly.

# app/worker.py
from arq import cron
from arq.connections import RedisSettings


async def sync_invoice(ctx: dict, invoice_id: str) -> dict:
    # ctx["job_try"] starts at 1 and increments on every retry — log it.
    pool = ctx["db"]                      # placed on ctx by the startup hook below
    await push_to_accounting(pool, invoice_id)
    return {"synced": invoice_id, "attempt": ctx["job_try"]}


async def nightly_cleanup(ctx: dict) -> None:
    await purge_expired_tokens(ctx["db"])


async def startup(ctx: dict) -> None:
    # The worker is its own process: it needs its OWN pool, sized in its own budget.
    ctx["db"] = await create_async_pool(settings.database_url, max_size=5)


async def shutdown(ctx: dict) -> None:
    await ctx["db"].close()


class WorkerSettings:
    functions = [sync_invoice]
    cron_jobs = [cron(nightly_cleanup, hour=3, minute=0)]
    redis_settings = RedisSettings.from_dsn(settings.redis_url)
    on_startup = startup
    on_shutdown = shutdown
    max_tries = 5                # attempts, not retries: 5 means 4 retries after the first try
    job_timeout = 300            # seconds; a job over this is cancelled and counted as a failure

The FastAPI side owns the pool and nothing else:

# app/main.py
from contextlib import asynccontextmanager

from arq import create_pool
from arq.connections import RedisSettings
from fastapi import FastAPI, HTTPException, Request

QUEUEABLE = {"sync_invoice", "nightly_cleanup"}     # names the worker will recognise


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.arq = await create_pool(RedisSettings.from_dsn(settings.redis_url))
    try:
        yield
    finally:
        await app.state.arq.close()


@app.post("/invoices/{invoice_id}/sync", status_code=202)
async def sync(invoice_id: str, request: Request) -> dict:
    # A natural key as the job id: the same invoice cannot be queued twice concurrently.
    job = await request.app.state.arq.enqueue_job(
        "sync_invoice", invoice_id, _job_id=f"sync:{invoice_id}"
    )
    if job is None:
        return {"status": "already_queued", "invoice_id": invoice_id}
    return {"status": "queued", "job_id": job.job_id, "invoice_id": invoice_id}

Note the 202 and the branch on None. Returning 200 {"status": "queued"} for both cases throws away the one signal ARQ gives you about duplicate work, and clients that retry on a network blip have no way to tell whether their second attempt did anything.

The enqueue side, executed

The following transcript is a real run of an app with exactly that route shape, against an in-process store that implements the slice of ARQ's pool contract the web process touches — enqueue_job with _job_id, refusal of a duplicate id, and encoding of arguments before storage. There is no Redis behind it and the page does not claim there is.

$ POST /invoices/inv-42/sync
200 OK
{
  "invoice_id": "inv-42",
  "enqueued": true,
  "job_id": "sync:inv-42",
  "note": "queued"
}

$ POST /invoices/inv-42/sync
200 OK
{
  "invoice_id": "inv-42",
  "enqueued": false,
  "job_id": null,
  "note": "duplicate collapsed into the in-flight job"
}

$ POST /invoices/inv-43/sync
200 OK
{
  "invoice_id": "inv-43",
  "enqueued": true,
  "job_id": "sync:inv-43",
  "note": "queued"
}

$ POST /invoices/inv-44/sync-object
422 Unprocessable Entity
{
  "detail": "TypeError: Object of type Invoice is not JSON serializable"
}

$ POST /enqueue-unknown
422 Unprocessable Entity
{
  "detail": "ValueError: unknown function 'snyc_invoice'"
}

The second request is the one to look at. The same invoice, the same job id, and no job created — the route can tell the client it was already queued because enqueue_job returned nothing. The fourth request is the serialization boundary made visible: passing the model object instead of its id fails inside the handler, which is where you want it, with a message that names the offending type.

The fifth is the string-name problem. snyc_invoice is a plausible typo that no static check would catch if it went straight to Redis; here the enqueue wrapper validates the name against the set the worker knows and rejects it at the API.

Inspecting what was stored shows what actually crosses the boundary:

$ GET /queue
200 OK
{
  "depth": 2,
  "job_ids": [
    "sync:inv-42",
    "sync:inv-43"
  ],
  "payloads": {
    "sync:inv-42": "{\"function\": \"sync_invoice\", \"args\": [\"inv-42\"]}",
    "sync:inv-43": "{\"function\": \"sync_invoice\", \"args\": [\"inv-43\"]}"
  }
}

A function name and a list of primitives. Nothing else. Once you have seen the payload it becomes obvious why passing an ORM row cannot work and why the task has to re-load the object itself.

The worker side, executed locally

The worker daemon cannot run here, but its calling convention can. Decoding each stored payload and invoking the same task callables with an ARQ-shaped ctx dict exercises the part people get wrong — the leading context argument and the job_try counter:

$ POST /drain
200 OK
{
  "results": [
    {
      "synced": "inv-42",
      "attempt": 1
    },
    {
      "synced": "inv-43",
      "attempt": 1
    }
  ],
  "task_side_effects": [
    {
      "function": "sync_invoice",
      "invoice_id": "inv-42",
      "attempt": 1,
      "worker": "worker-a"
    },
    {
      "function": "sync_invoice",
      "invoice_id": "inv-43",
      "attempt": 1,
      "worker": "worker-a"
    }
  ]
}

The tasks are the same functions the real worker would import. What a real worker adds on top is scheduling, the retry loop that increments job_try, the timeout enforcement, and result storage — behaviour of the ARQ daemon, described here and deliberately not simulated.

Deploying the Worker

The worker is a process, not a thread and not a subprocess of the API. Run it from the same image with a different command so code and dependencies stay in lockstep:

# api container
CMD ["uvicorn", "app.main:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000"]

# worker container: same image, different entrypoint
CMD ["arq", "app.worker.WorkerSettings"]

Sharing an image and splitting the command is what keeps functions in WorkerSettings consistent with the names your routes enqueue. If the API deploys and the worker does not, jobs referring to a newly added function pile up unhandled — so roll the worker first when adding a task, and roll the API first when removing one.

Because the worker builds its own database pool in on_startup, its connections are additional to the web tier's. Size both against the database's limit together; the failure mode when you do not is covered in fixing asyncpg pool exhaustion.

Verification

The enqueue path is unit-testable without Redis if you keep the pool behind app.state and override it, which is the same technique used in overriding dependencies in tests:

async def test_duplicate_submission_collapses(app_with_fake_pool):
    first = await client.post("/invoices/inv-1/sync")
    second = await client.post("/invoices/inv-1/sync")
    assert first.json()["status"] == "queued"
    assert second.json()["status"] == "already_queued"


async def test_arguments_are_encodable(app_with_fake_pool):
    # Fails loudly at enqueue time rather than on a worker at 3am.
    with pytest.raises(TypeError):
        await pool.enqueue_job("sync_invoice", Invoice(id="inv-1"))

The task bodies deserve their own tests, called directly with a ctx dict you construct — {"job_try": 1, "db": test_pool} is enough. That is a plain async function test with no queue involved, and it covers the logic that actually matters.

In production, the two signals worth alerting on are queue depth and the age of the oldest queued job. Depth alone is misleading, because a healthy burst looks identical to a stalled worker for the first minute; oldest-job age separates them. arq --check reports worker health for a liveness probe, and forwarding the request's correlation id into the job as an ordinary argument makes a job traceable back to the request that created it — see correlating logs, traces and errors.

Trade-offs and When Not To Use ARQ

Redis is the only broker, and it is now a durability dependency. ARQ does not speak AMQP. If your queue must survive the loss of the Redis node, you need Redis persistence configured deliberately — and Redis persistence is weaker than a real message broker's. For work where losing a job is a financial event, that constraint matters more than ARQ's ergonomics.

The ecosystem is small. Celery has integrations, monitoring dashboards, and a decade of answered questions. With ARQ you will write your own dead-letter handling and your own dashboard queries. That is a fair trade when the task surface is a handful of functions, and a poor one when a team of twenty needs a shared operational vocabulary.

Retries are at-least-once, so idempotency is not optional. A job that timed out may still have completed its side effect. max_tries will run it again anyway. Every task needs a natural key and an upsert or a guard, which is the whole subject of retry and idempotency for tasks.

Cron jobs run per worker deployment, not per replica. ARQ coordinates through Redis so a cron job fires once across the pool, but that also means a cron schedule is tied to your worker's uptime. If the whole worker deployment is down at 03:00, the nightly job does not run late — it does not run.

If the work is short and losing it is acceptable, this is all overhead. A durable queue costs you a second deployable, a second connection budget, and a second thing to page on. BackgroundTasks remains the right answer for a best-effort cache warm, with the caveat documented in when BackgroundTasks silently fails.

FAQ

How does FastAPI enqueue a job that an ARQ worker runs? The FastAPI process opens an ARQ Redis pool at startup and calls enqueue_job on it from a route, which writes the encoded job into Redis. A separately running ARQ worker process pops the job, decodes it, and calls the matching function by name. The two processes share Redis and nothing else.

How do I prevent duplicate ARQ jobs from a double-submitted request? Pass an explicit _job_id derived from the operation's natural key. ARQ refuses to enqueue a job id that already exists and returns None instead of a Job, so a second identical submission collapses into the first. Combine that with an idempotent task body, because the deduplication window ends once the job completes and its result expires.

Why does enqueue_job fail with a serialization error? Because arguments are encoded before they are written to Redis, and an ORM row, an open client, or anything holding a lambda cannot be encoded. Pass the identifier and re-load the object inside the task, where a worker-owned database session is available.

Can ARQ run scheduled or recurring jobs? Yes. Add cron entries to WorkerSettings using the cron helper with the function and its schedule. The worker runs them itself, so there is no separate scheduler service to deploy and keep alive, which is a real operational saving compared with Celery beat.

Does the worker share the FastAPI app's database pool? No. The worker is a separate process with its own event loop, so it builds its own engine and pool in its startup hook. Budget those connections alongside the web tier or a busy queue and a busy API together will exhaust the database.

What is the ctx argument in an ARQ task? It is a dict the worker passes as the first positional argument to every task, carrying the job id, the current attempt number as job_try, the worker's identity, and anything the worker's own startup hook stored on it, such as a database pool or an HTTP client.

Related Reading

• Up to the topic: Background Task Processing for where queued work sits relative to in-process tasks.
• Choosing this at all: FastAPI BackgroundTasks vs Celery vs ARQ compares durability and operational cost.
• The rule at-least-once delivery imposes: Retry and idempotency for tasks.
• Budgeting the worker's connections: Fixing asyncpg pool exhaustion.
• Tracing a job back to its request: Correlating logs, traces and errors.