FastAPI + Celery + Redis — Distributed Task Queue Tutorial

🎯 What You Will Build

A FastAPI endpoint that dispatches heavy work (report generation, emails, image processing) to a Celery worker via Redis, returns instantly with a task id, and lets the client poll for the result — a real distributed task queue.

The broker → worker → result-backend model
When to reach for Celery instead of BackgroundTasks
Defining tasks and dispatching with .delay()
Tracking task status/results with AsyncResult
Automatic retries on failure
Running the worker + full Docker Compose stack

Prerequisites: Python 3.11+, a running Redis, and comfort with FastAPI routes. If you've used BackgroundTasks before (see fastapi-background-tasks), this is the next step up.

fastapi-celery/
├── .env
├── requirements.txt
├── docker-compose.yml
└── app/
    ├── __init__.py
    ├── main.py         # FastAPI routes
    ├── celery_app.py   # the Celery instance
    └── tasks.py        # task definitions

The Mental Model: Broker, Worker, Result Backend

Celery splits work across processes. Your FastAPI app is the producer: it puts a job on a queue and moves on. A separate worker process pulls jobs off the queue and runs them. Redis plays two roles — the broker (the queue) and the result backend (where results are stored).

FastAPI (producer)                Worker (consumer)
      |                                  ^
      |  task.delay(args)                | pulls job
      v                                  |
   Redis broker  ---- job queued ----->  runs task
      ^                                  |
      |  <---- result stored ----------  |
   Redis result backend  <--------------- returns result

Client:  POST /work  ->  {task_id}  ->  GET /work/{task_id}  ->  status/result

💡

Ang importante dito: the API and the worker are separate processes. You can restart, scale, or crash the worker without taking down the API — and add more workers to process the queue faster.

Celery vs BackgroundTasks: When Do You Need This?

FastAPI's built-in BackgroundTasks runs work inside the same process after the response. That's fine for quick, fire-and-forget jobs — but it dies when the server restarts and competes with request handling. Celery is for real, durable, heavy work.

Need	BackgroundTasks	Celery + Redis
Runs in	Same web process	Separate worker process(es)
Survives app restart	❌ lost	✅ queued in Redis
Retries on failure	❌ manual	✅ built-in
Track status / get result	❌ no	✅ AsyncResult
Scheduled / periodic jobs	❌ no	✅ Celery Beat
Scale independently	❌ no	✅ add more workers
Best for	Quick side effects (log, cheap email)	Heavy/critical jobs (reports, media, batch)

⚠️

Rule of thumb: if losing the job on a restart would be a bug, or the job takes more than a second or two, use Celery. Otherwise BackgroundTasks is simpler.

Setup + .env

requirements.txt

fastapi==0.116.1
uvicorn[standard]==0.30.6
celery==5.4.0
redis==5.2.1
pydantic-settings==2.6.1

.env

# broker = the job queue, backend = where results are stored
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/1

💡

Using two Redis databases (/0 for the broker, /1 for results) keeps queued jobs and stored results from colliding. It also makes it easy to flush one without the other.

The Celery Instance

Create one Celery app that both the API and the worker import. This is the shared configuration.

app/celery_app.py

from celery import Celery
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    CELERY_BROKER_URL: str = "redis://localhost:6379/0"
    CELERY_RESULT_BACKEND: str = "redis://localhost:6379/1"
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


settings = Settings()

celery_app = Celery(
    "worker",
    broker=settings.CELERY_BROKER_URL,
    backend=settings.CELERY_RESULT_BACKEND,
    include=["app.tasks"],   # modules the worker should import
)

celery_app.conf.update(
    task_track_started=True,        # report a "STARTED" state, not just PENDING->SUCCESS
    task_time_limit=300,            # hard kill a task after 5 min
    result_expires=3600,            # drop results from Redis after 1 hour
)

💡

include=["app.tasks"] tells the worker where to find your tasks. Forget it and the worker starts but reports Received unregistered task when a job arrives.

Defining Tasks

A task is a normal function decorated with @celery_app.task. Keep tasks small, serializable (JSON-friendly args), and idempotent where possible.

app/tasks.py

import time

from app.celery_app import celery_app


@celery_app.task(name="generate_report")
def generate_report(report_id: int) -> dict:
    # Pretend this is a heavy aggregation that takes ~5 seconds.
    time.sleep(5)
    return {"report_id": report_id, "total_sales": 128_400, "orders": 512}


@celery_app.task(name="send_email")
def send_email(to: str, subject: str) -> dict:
    time.sleep(2)  # pretend SMTP call
    return {"to": to, "subject": subject, "status": "sent"}

⚠️

Pass ids, not objects. Task arguments are serialized to Redis. Send a user_id, not a SQLAlchemy User or a DB session — the worker is a different process and re-loads what it needs.

Dispatching Tasks from FastAPI

In your route, call .delay(...) (a shortcut for .apply_async). It enqueues the job and returns immediately with a task id — the endpoint responds in milliseconds even though the work takes seconds.

app/main.py

from fastapi import FastAPI, status
from pydantic import BaseModel

from app.tasks import generate_report, send_email

app = FastAPI(title="FastAPI Celery", version="0.1.0")


class EmailRequest(BaseModel):
    to: str
    subject: str


@app.post("/reports/{report_id}", status_code=status.HTTP_202_ACCEPTED)
def start_report(report_id: int):
    task = generate_report.delay(report_id)
    return {"task_id": task.id, "status": "queued"}


@app.post("/emails", status_code=status.HTTP_202_ACCEPTED)
def queue_email(data: EmailRequest):
    task = send_email.delay(data.to, data.subject)
    return {"task_id": task.id, "status": "queued"}

💡

Why 202 Accepted? The correct status for "I've accepted your request and will process it later." It tells the client the work isn't done yet — poll the status endpoint for the result.

Tracking Task Status & Results

Given a task id, AsyncResult reads the current state and result from the Redis backend. Expose it as a status endpoint the client can poll.

app/main.py

from celery.result import AsyncResult

from app.celery_app import celery_app


@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    response = {"task_id": task_id, "state": result.state}

    if result.successful():
        response["result"] = result.result
    elif result.failed():
        response["error"] = str(result.result)  # the exception message

    return response

State	Meaning
PENDING	Unknown / not yet picked up by a worker
STARTED	A worker is running it (needs task_track_started)
RETRY	Failed and scheduled to retry
SUCCESS	Finished; result is available
FAILURE	Raised an exception; result holds the error

⚠️

Gotcha: an unknown task id also shows PENDING — Celery can't tell "not started yet" from "never existed." Don't treat PENDING as proof the job is queued.

Retries & Failure Handling

The biggest reason to use Celery: automatic retries. If a task hits a transient error (a flaky API, a timeout), Celery can retry it with backoff instead of losing the job.

app/tasks.py

import random

from app.celery_app import celery_app


@celery_app.task(
    name="charge_payment",
    bind=True,                       # gives us "self" to call self.retry()
    autoretry_for=(ConnectionError,),  # auto-retry on these exceptions
    retry_backoff=True,              # 1s, 2s, 4s, ... exponential backoff
    retry_kwargs={"max_retries": 5},
    acks_late=True,                  # re-queue if the worker dies mid-task
)
def charge_payment(self, order_id: int) -> dict:
    if random.random() < 0.5:
        raise ConnectionError("payment gateway timed out")
    return {"order_id": order_id, "charged": True}

💡

acks_late=True means the job is only acknowledged after it finishes. If the worker is killed mid-task, Redis keeps the job and another worker picks it up — critical for money/data tasks.

⚠️

Retries need idempotency. A task that may run more than once must be safe to run more than once — check "already charged?" before charging, or you'll double-charge on a retry.

Running It (Worker + Docker Compose)

You now run two processes: the API and the worker. Start Redis, then each process in its own terminal:

run-local.sh

# terminal 1: Redis
docker run -d -p 6379:6379 redis

# terminal 2: the FastAPI app
uvicorn app.main:app --reload

# terminal 3: the Celery worker
celery -A app.celery_app worker --loglevel=info

For a reproducible stack, wire all three together with Docker Compose:

docker-compose.yml

services:
  redis:
    image: redis:7
    ports:
      - "6379:6379"

  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000
    ports:
      - "8000:8000"
    environment:
      CELERY_BROKER_URL: redis://redis:6379/0
      CELERY_RESULT_BACKEND: redis://redis:6379/1
    depends_on:
      - redis

  worker:
    build: .
    command: celery -A app.celery_app worker --loglevel=info
    environment:
      CELERY_BROKER_URL: redis://redis:6379/0
      CELERY_RESULT_BACKEND: redis://redis:6379/1
    depends_on:
      - redis

💡

Note the broker URL is redis://redis:6379/0 inside Compose — redis is the service name, not localhost. Containers talk to each other by service name on the Compose network.

Try It (curl)

1Queue a job — get a task id back instantly

$ curl -s -X POST http://127.0.0.1:8000/reports/7
{"task_id":"b1c2...","status":"queued"}   # returns immediately, not after 5s

2Poll the status until it's SUCCESS

# right away
$ curl -s http://127.0.0.1:8000/tasks/b1c2...
{"task_id":"b1c2...","state":"STARTED"}

# ~5 seconds later
$ curl -s http://127.0.0.1:8000/tasks/b1c2...
{"task_id":"b1c2...","state":"SUCCESS","result":{"report_id":7,"total_sales":128400,"orders":512}}

# meanwhile, the worker terminal logs:
# [INFO] Task generate_report[b1c2...] received
# [INFO] Task generate_report[b1c2...] succeeded in 5.01s

✅

You did it: the API responded in milliseconds, the worker did the 5-second job in the background, and the client polled for the result. That's a distributed task queue.

Common Mistakes (and Fixes)

Mistake	Symptom	Fix
Not running a worker	Tasks stay PENDING forever	Start `celery -A app.celery_app worker`
Missing `include=[...]`	Received unregistered task	Register the task module on the Celery app
Passing DB objects/sessions to a task	Serialization error / stale data	Pass ids; reload inside the task
Using `localhost` in Docker Compose	Worker can't reach the broker	Use the service name (`redis`)
Retrying a non-idempotent task	Double charges / duplicate emails	Make tasks safe to run more than once
Calling `result.get()` inside a request	The endpoint blocks — defeats the whole point	Return the task id; poll a status endpoint

⚠️

Never block on the result in a request handler. Calling .get() makes FastAPI wait for the task to finish, turning your async win back into a slow, blocking request. Always hand back the id.

What's Next

✅

🚀 Recommended next reads:

fastapi-background-tasks — the simpler in-process option, and when it's enough
fastapi-redis-caching — you already have Redis; cache hot reads too
fastapi-docker — package this API + worker stack into images
fastapi-async-sqlalchemy — have tasks reload records by id from the database

Recap: FastAPI enqueues with .delay(), Redis holds the queue and results, and a separate Celery worker does the heavy lifting. Return a task id, poll AsyncResult, lean on built-in retries, and keep tasks idempotent. Now long jobs never block a request again.

FastAPI + Celery + Redis (Distributed Task Queue)

🎯 What You Will Build

The Mental Model: Broker, Worker, Result Backend

Celery vs BackgroundTasks: When Do You Need This?

Setup + .env

The Celery Instance

Defining Tasks

Dispatching Tasks from FastAPI

Tracking Task Status & Results

Retries & Failure Handling

Running It (Worker + Docker Compose)

Try It (curl)

Common Mistakes (and Fixes)

What's Next

About the Author

Thirdy Gayares

🎯 What You Will Build

1The Mental Model: Broker, Worker, Result Backend

2Celery vs BackgroundTasks: When Do You Need This?

3Setup + .env

4The Celery Instance

5Defining Tasks

6Dispatching Tasks from FastAPI

7Tracking Task Status & Results

8Retries & Failure Handling

9Running It (Worker + Docker Compose)

10Try It (curl)

11Common Mistakes (and Fixes)

12What's Next

About the Author

Thirdy Gayares

The Mental Model: Broker, Worker, Result Backend

Celery vs BackgroundTasks: When Do You Need This?

Setup + .env

The Celery Instance

Defining Tasks

Dispatching Tasks from FastAPI

Tracking Task Status & Results

Retries & Failure Handling

Running It (Worker + Docker Compose)

Try It (curl)

Common Mistakes (and Fixes)

What's Next