Running Background Jobs with Celery to Keep API Latency Low

When a request triggers slow work — sending a welcome email, rendering a PDF report, fanning a webhook out to a dozen subscribers — doing it inline blocks the response and pushes your p95 latency through the roof. The fix is to hand that work to a worker process and return immediately. This guide shows how to wire Celery into a FastAPI service so routes enqueue work in milliseconds and a separate worker pool grinds through it. Part of the Scaling and Operating Production Python APIs guide.

The pattern is simple: your API publishes a message to a broker, a worker pool pulls messages off that broker, runs the task, and writes the outcome to a result backend. Your route never waits. That single architectural move — splitting "accept the work" from "do the work" — is the difference between a route that answers in 8 ms and one that answers in 8 seconds because it is sitting on an SMTP handshake. Once you internalise it, you start seeing background-job candidates everywhere: anything that talks to a third party, anything CPU-heavy, anything you would be embarrassed to make a paying customer wait for.

The commercial framing matters here. Slow synchronous routes do not just annoy users; they burn money. A web worker tied up for six seconds sending an email is a web worker that cannot serve other paying requests, so you scale your API tier to cover work that has nothing to do with serving HTTP. Moving that work to a cheap background pool lets you run fewer, smaller web dynos and keep tail latency flat under load — the same discipline behind response caching with Redis.

Prerequisites

You need Python 3.11+, a running broker, and two packages. Redis doubles as both broker and result backend, which is why most builders reach for it first — if you already run Redis for caching, you have zero new infrastructure to provision, monitor, or pay for.

pip install "celery[redis]>=5.4" fastapi uvicorn

Set two environment variables. Keep them in your secret manager, never in code:

export CELERY_BROKER_URL="redis://localhost:6379/0"
export CELERY_RESULT_BACKEND="redis://localhost:6379/1"

Using separate Redis logical databases (/0 for the broker, /1 for results) keeps queued messages and stored results from colliding when you flush one of them. In production, point both at a managed Redis instance over TLS (rediss://) rather than a Redis you run yourself — the few dollars a month a managed instance costs is far cheaper than the afternoon you will lose to a self-hosted eviction storm that silently drops queued jobs. A broker that loses messages is worse than no broker at all, because it fails silently.

One decision to make up front: do you actually need a result backend? If your tasks are fire-and-forget — send this email, post this webhook — you never read the return value, so storing it just fills Redis with garbage. Drop CELERY_RESULT_BACKEND entirely and set ignore_result=True. Keep the backend only when a client genuinely polls for task status or you chain tasks together and one needs another's output.

Step 1: Define a Celery app from the environment

Create worker.py. The app reads its broker and backend from os.getenv so the same code runs unchanged in dev, staging, and production — the environment differs, the code does not.

import os
from celery import Celery

celery_app = Celery(
    "api_jobs",
    broker=os.getenv("CELERY_BROKER_URL"),
    backend=os.getenv("CELERY_RESULT_BACKEND"),
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_reject_on_worker_lost=True,
    result_expires=3600,
    broker_transport_options={"visibility_timeout": 3600},
)

Pinning json serialization avoids the pickle security hole: with pickle enabled, anyone who can write to your broker can execute arbitrary code inside your worker, so a compromised or shared Redis becomes remote code execution. JSON also forces you to keep task arguments to primitives, which is a healthy constraint.

task_acks_late=True and worker_prefetch_multiplier=1 together mean a task is only acknowledged after it finishes, so a crashed worker hands its task back to the queue instead of silently losing it. task_reject_on_worker_lost=True closes the matching gap where a worker is killed by the OS (an OOM kill or a SIGKILL during a deploy) rather than raising an exception — without it, that task can be marked failed rather than requeued. The visibility_timeout tells Redis how long a claimed-but-unacknowledged message stays hidden before it becomes visible again; set it comfortably longer than your slowest task, or a long task will be redelivered to a second worker while the first is still running it.

Step 2: Write a task

A task is a normal function decorated with @celery_app.task. Keep the signature JSON-friendly — pass an ID, not an ORM object. The task re-fetches whatever it needs from the database itself, which also guarantees it works on the freshest row rather than a snapshot captured at enqueue time.

import smtplib
from email.message import EmailMessage

@celery_app.task(name="jobs.send_welcome_email")
def send_welcome_email(user_email: str, display_name: str) -> str:
    msg = EmailMessage()
    msg["To"] = user_email
    msg["From"] = os.getenv("SMTP_FROM")
    msg["Subject"] = "Welcome aboard"
    msg.set_content(f"Hi {display_name}, thanks for signing up.")

    with smtplib.SMTP(os.getenv("SMTP_HOST"), int(os.getenv("SMTP_PORT", "587"))) as s:
        s.starttls()
        s.login(os.getenv("SMTP_USER"), os.getenv("SMTP_PASSWORD"))
        s.send_message(msg)
    return user_email

Always name your tasks explicitly with name="jobs.send_welcome_email". If you let Celery auto-generate the name from the module path, renaming or moving the file changes the task's identity, and any messages already sitting in the queue under the old name become unroutable — the worker sees a task it does not recognise and dead-letters it. An explicit, stable name decouples the wire format from your file layout, which matters the day you refactor worker.py.

Give tasks a hard ceiling too. A single hung SMTP connection or a runaway loop can pin a worker slot forever, and enough of those starve the whole pool. Set task_time_limit (a hard kill) and task_soft_time_limit (a catchable SoftTimeLimitExceeded you can clean up around) either globally in the config or per task with @celery_app.task(time_limit=30, soft_time_limit=25). A task with no timeout is a latent outage.

Step 3: Enqueue from a FastAPI route with .delay()

In your route, call .delay() to publish the task and return at once. The HTTP response goes out in milliseconds; the email sends in the background.

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from worker import send_welcome_email

app = FastAPI()

class SignupIn(BaseModel):
    email: EmailStr
    name: str

@app.post("/signup", status_code=202)
async def signup(body: SignupIn) -> dict:
    task = send_welcome_email.delay(body.email, body.name)
    return {"status": "queued", "task_id": task.id}

Returning 202 Accepted with the task_id is the honest status code: you accepted the work but haven't finished it. The client can poll a status route later using that ID. Crucially, calling .delay() is non-blocking — it just writes a small message to Redis, so it never stalls the async event loop, even though send_welcome_email itself is a synchronous function.

.delay() is the ergonomic shorthand; reach for .apply_async() when you need control. It exposes the levers that matter in production: countdown=30 to defer a task by 30 seconds, eta=... to run it at a specific time, queue="emails" to route it to a dedicated queue, and priority=... on brokers that honour it. Routing is the one most builders adopt first — send transactional email to a high-priority emails queue and heavy report generation to a separate reports queue, then run workers dedicated to each so a backlog of nightly reports never delays a password-reset email a customer is staring at.

send_welcome_email.apply_async(
    args=[body.email, body.name],
    queue="emails",
    countdown=5,
)

Expose a status route so clients can poll without holding the connection open. Read the result by ID and translate Celery's internal state into a plain JSON shape — never leak the raw exception to the caller.

from celery.result import AsyncResult

@app.get("/tasks/{task_id}")
async def task_status(task_id: str) -> dict:
    res = AsyncResult(task_id, app=send_welcome_email.app)
    return {"task_id": task_id, "state": res.state, "ready": res.ready()}

Step 4: Add retries and idempotency

Network calls fail. Bind the task and retry with backoff so a flaky SMTP server or webhook endpoint doesn't drop work on the floor. This is the same fan-out pattern you'd use after processing webhooks with Python — accept the inbound hook fast, then deliver to each subscriber from a retrying task. It mirrors the client-side discipline of retrying failed HTTP requests with tenacity, except the queue is now your retry engine.

import httpx

@celery_app.task(
    bind=True,
    name="jobs.deliver_webhook",
    max_retries=5,
    acks_late=True,
    retry_backoff=True,
    retry_jitter=True,
)
def deliver_webhook(self, url: str, payload: dict, idempotency_key: str) -> int:
    try:
        resp = httpx.post(
            url,
            json=payload,
            headers={"Idempotency-Key": idempotency_key},
            timeout=10.0,
        )
        resp.raise_for_status()
        return resp.status_code
    except httpx.HTTPError as exc:
        # Exponential backoff: 2s, 4s, 8s, 16s, 32s (capped, with jitter)
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

The doubling delay matters. If a downstream service is briefly overloaded, hammering it every second makes the outage worse and can trip its rate limiter, turning a 30-second blip into a self-inflicted 429 storm. Backing off — 2s, 4s, 8s, 16s — gives the dependency room to recover. Add retry_jitter=True so a thundering herd of tasks that all failed at the same instant do not all retry at the same instant; jitter smears them across the window. The figure below traces a single message that fails four times before succeeding on the fifth attempt.

The idempotency_key matters because acks_late can replay a task if a worker dies mid-flight, and the retry loop above deliberately re-sends the same request. Send the same key every time so the receiver de-duplicates instead of double-charging or double-sending. If you own the receiving end, the companion pattern is building an idempotent webhook receiver that records processed keys and short-circuits repeats. After max_retries is exhausted, do not let the task vanish — route it to a dead-letter queue or write the failed payload to a table you can inspect and replay, so a permanently broken endpoint leaves an audit trail instead of a silent gap.

Step 5: Schedule recurring work with Celery beat

For report generation that runs nightly, add a beat schedule. Beat is a scheduler that publishes tasks to the broker on a timer; the workers run them like any other task. Beat itself does no work — it only enqueues — which is exactly why it stays lightweight even when the jobs it triggers are heavy.

from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "nightly-usage-report": {
        "task": "jobs.generate_usage_report",
        "schedule": crontab(hour=2, minute=0),
    },
    "hourly-usage-rollup": {
        "task": "jobs.rollup_usage",
        "schedule": crontab(minute=0),
    },
}

@celery_app.task(name="jobs.generate_usage_report")
def generate_usage_report() -> str:
    # Aggregate yesterday's metered usage, render a PDF, upload it.
    return "report-ok"

Run beat as its own process alongside the workers:

celery -A worker.celery_app worker --concurrency=4 --loglevel=info
celery -A worker.celery_app beat --loglevel=info

The one rule that trips people up: run exactly one beat process, ever. Two beat schedulers pointed at the same broker each fire the schedule, so your "nightly" report runs twice and your metered-usage numbers double. Workers scale horizontally; beat is a singleton. When you deploy on a platform that can spin up multiple instances, pin beat to a single instance or use a lock so a rolling deploy never briefly runs two. The figure shows how a handful of schedules with different cadences interleave over a single day.

Recurring usage rollups feed naturally into metered API billing, where the nightly job tallies each customer's calls before you report them to Stripe. If your scheduling needs are simpler than a full Celery deployment — a couple of cron-like jobs and nothing else — weigh the lighter option first; the trade-offs are laid out in APScheduler vs Celery beat, and the broader picture in scheduling data pipelines with cron.

Configuration reference

The numbers that actually move your bill are --concurrency and the worker count, covered next. Everything above is correctness and hygiene; those two are cost.

Sizing the worker pool

Concurrency is where builders either save money or waste it. A worker process with --concurrency=8 runs eight tasks at once using the default prefork pool — eight OS subprocesses. For I/O-bound work (email, HTTP fan-out, waiting on a third-party API) those subprocesses spend almost all their time blocked on a socket, so you can pack many onto a small box and throughput climbs nearly linearly until you saturate the CPU or the downstream service. For CPU-bound work (PDF rendering, image resizing, number crunching) each task actually wants a core, so concurrency above your core count just context-switches and thrashes; there you scale by adding worker machines, not by raising the number.

The chart below is a realistic profile for an I/O-bound task averaging about 0.4 seconds of mostly-waiting work. Throughput roughly doubles from concurrency 4 to 8 to 16, then flattens as the box's CPU becomes the bottleneck — past 16, you are buying context-switch overhead, not throughput.

The practical rule: size the pool to drain your queue faster than it fills during peak, then leave a little headroom. Watch queue depth, not CPU — if the number of waiting messages trends up over a busy hour, you are under-provisioned and latency-sensitive jobs are aging in the queue. For genuinely async I/O you can also run the worker with --pool=gevent or --pool=eventlet and push concurrency into the hundreds on a single box, since green threads are far cheaper than subprocesses. Once the worker fleet is stable, containerise it the same way you would the API — containerizing Python APIs with Docker applies unchanged, just with the celery ... worker command as the entrypoint instead of Uvicorn.

Gotchas and failure modes

1. Blocking the event loop. Never call a task's .get() inside an async route to wait for the result — that blocks the loop and defeats the point. Return the task_id and poll, or fire a webhook callback when the task finishes.
2. Non-serializable arguments. Passing a SQLAlchemy object or a raw datetime without a serializer raises an encoding error on enqueue. Pass primitive IDs and re-fetch inside the task, opening a fresh session with async SQLAlchemy rather than reusing the request's session.
3. Lost tasks without acks_late. With the default early ack, a task is marked done the instant a worker picks it up. If that worker is killed (a deploy, an OOM) the work vanishes. Turn on acks_late for anything that matters, and pair it with idempotency.
4. Result backend bloat. Every task return value is stored by default. On a busy API this fills Redis fast and can trigger evictions that also drop queued messages. Set result_expires, or set ignore_result=True on fire-and-forget tasks.
5. Non-idempotent tasks. Because late-acked tasks can re-run, a task that charges a card or sends an email must be safe to execute twice. Guard with an idempotency key or an "already processed" check.
6. Sharing a DB connection across the fork. The prefork pool forks after the module imports, so a database engine created at import time gets copied into every child and its sockets collide. Create engines and connection pools lazily inside the task or on the worker-ready signal, not at module scope.

Verification

Start a worker and watch the logs. A healthy enqueue shows the task received and succeeded:

[INFO/MainProcess] Task jobs.send_welcome_email[3f9c...] received
[INFO/ForkPoolWorker-1] Task jobs.send_welcome_email[3f9c...] succeeded in 0.42s

Fire a request and confirm the route returns immediately:

curl -s -X POST localhost:8000/signup \
  -H "content-type: application/json" \
  -d '{"email":"a@b.com","name":"Ada"}'
# {"status":"queued","task_id":"3f9c..."}

For a live dashboard of queues, workers, and task history, run Flower:

pip install flower
celery -A worker.celery_app flower --port=5555

It serves a UI on localhost:5555 showing per-task runtime, retries, and failures — the fastest way to catch a backed-up queue. Flower is fine for eyeballing, but for real alerting wire the worker into your existing observability. Celery emits per-task events and exposes queue length through the broker, both of which slot straight into monitoring and logging your Python APIs — emit a structured log line per task with its duration and retry count via structlog, then alert on queue depth crossing a threshold rather than waiting for a customer to notice their email never arrived. Cover the tasks themselves in your test suite too; testing Python APIs with pytest shows how to run tasks eagerly (task_always_eager=True) so a task's logic is asserted inline without a live broker.

Cost and performance note

Workers are the line item. A single worker process with --concurrency=8 on a small instance comfortably clears tens of thousands of I/O-bound tasks an hour, and that box rents for a few dollars a month. At a million background jobs a month — a welcome email, a couple of webhook deliveries, and a nightly rollup per customer across a healthy user base — you are looking at roughly two to four dollars of worker compute if you keep concurrency matched to the workload and the pool busy rather than idle. The expensive mistake is running background work on your web tier: a Uvicorn dyno sized for request latency is far pricier per unit of throughput than a cheap worker box, and every second it spends on a background job is a second it is not serving the requests you scaled it for.

Because the worker pool is decoupled, you can scale it independently and even down to zero between beat runs on a platform that bills per-second. Pair this with response caching so the synchronous side of your API stays cheap while the worker pool absorbs the slow work, and deploy the two tiers separately so a worker rollout never risks request-serving uptime — the discipline in zero-downtime deploys for Python APIs applies to draining a worker gracefully before you replace it, so in-flight tasks finish instead of dying mid-charge.

FAQ

How much does this cost to run at 1M background jobs a month? Two to four dollars of worker compute if you keep concurrency matched to the workload and the pool busy rather than idle. A single small worker box with --concurrency=8 clears tens of thousands of I/O-bound tasks an hour. The costly error is running the work on your web tier, where compute is priced for request latency, not throughput — move it to a dedicated worker and the margin math changes entirely.

Do I need Celery if FastAPI already has BackgroundTasks?BackgroundTasks runs in the same process as your API, so a deploy or crash kills any in-flight work and there are no retries. Celery survives restarts, retries failures, and scales workers independently. Use BackgroundTasks for trivial fire-and-forget where losing the odd job is harmless; use Celery the moment the work must not be lost or must run somewhere other than your web process.

Can I run the worker on the same machine as the API? Yes for low volume — just run the celery worker command as a second process. As load grows, move workers to their own machines so a flood of background jobs can't starve your request-serving CPU, and so you can scale each tier on its own cost curve. That separation is also what lets you deploy workers without touching the API.

How do I avoid double-charging a customer when a task runs twice? Make tasks idempotent. Pass a deterministic idempotency key and check whether that key was already processed before doing the side effect. This is mandatory when acks_late is on, because a dead worker's task gets requeued and re-run. For payments specifically, lean on the provider's own idempotency support so a retried charge collapses into one.

Should I use Celery or arq for an async-native FastAPI app? It depends on your stack and how much operational machinery you want to run. Celery is battle-tested with rich scheduling and multi-broker support but runs synchronous workers; arq is tiny and async-native. See Celery vs RQ vs arq for the side-by-side before you commit.

Related

Same track:

• Celery vs RQ vs arq for Python background jobs — pick the right queue before you wire one in.
• Caching Python API Responses with Redis — reuse the same Redis you run as a broker to cut synchronous latency.
• Monitoring and Logging Python APIs — alert on queue depth and per-task duration instead of waiting for a complaint.
• Scaling and Operating Production Python APIs — the wider section this guide sits in.

Other tracks:

• Processing Webhooks with Python — the classic accept-fast, deliver-from-a-task pattern.
• How to Charge for API Access Using Stripe — turn nightly usage rollups into metered invoices.
• APScheduler vs Celery beat — when a full beat deployment is more than your schedule needs.