0Pricing
FastAPI Backend Development Bootcamp · Lesson

Wiring Celery Workers to a FastAPI App

Configure Celery with a Redis broker, define tasks, and dispatch jobs from endpoints with result backends.

Wiring Celery Workers to a FastAPI App is a free FastAPI Backend Development Bootcamp lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the FastAPI Backend Development Bootcamp learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Celery in a FastAPI App?

FastAPI request handlers must return quickly. Work like sending emails, generating reports, resizing images, or calling slow third-party APIs can take seconds, blocking your worker process and hurting throughput.

Celery is a distributed task queue. You push a job onto a queue, and a separate pool of worker processes runs it outside the request/response cycle.

  • Broker — the message transport that holds queued jobs (we use Redis).
  • Worker — long-running process that pulls and executes tasks.
  • Result backend — optional store for return values and task state.

The endpoint stays fast and just answers: "accepted, here is your job id."

Installing the Pieces

Install Celery with the Redis extra, plus a Redis server reachable from both your API and your workers.

  • celery[redis] pulls in Celery and the redis client.
  • Run Redis locally with Docker: docker run -p 6379:6379 redis.

Both the FastAPI process and the worker process import the same Celery application object, so they must share the same codebase and broker URL.

# requirements.txt
fastapi
uvicorn[standard]
celery[redis]
redis

# install
# pip install -r requirements.txt
# run redis: docker run -p 6379:6379 redis:7

Creating the Celery App

Define a single Celery application instance in its own module (commonly worker.py or celery_app.py). It needs a name, a broker URL, and a result backend URL.

  • broker — where tasks are enqueued (Redis DB 0).
  • backend — where results/state are stored (Redis DB 1, kept separate for clarity).

The first argument ("worker") becomes the default prefix for task names.

# celery_app.py
from celery import Celery

celery_app = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    enable_utc=True,
)

Defining Your First Task

A task is just a function decorated with @celery_app.task. When called normally it runs inline; when called with .delay() or .apply_async() it is serialized and pushed to the broker.

  • Arguments must be JSON-serializable (use ids and primitives, not ORM objects).
  • Give the task an explicit name so renaming the function later does not break queued messages.
# tasks.py
import time
from celery_app import celery_app

@celery_app.task(name="tasks.send_report")
def send_report(user_id: int, email: str) -> dict:
    # simulate slow work
    time.sleep(5)
    return {"user_id": user_id, "sent_to": email, "status": "done"}

Dispatching a Job from an Endpoint

Inside a FastAPI route, call .delay(...) to enqueue the task. This returns an AsyncResult immediately — it does not wait for the work to finish.

Respond with the task.id and HTTP 202 Accepted, signalling that the request was accepted for processing but is not yet complete.

Never .get() the result inside the request handler — that blocks the event loop until the job finishes, defeating the entire purpose.

# main.py
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from tasks import send_report

app = FastAPI()

class ReportRequest(BaseModel):
    user_id: int
    email: EmailStr

@app.post("/reports", status_code=202)
def create_report(req: ReportRequest):
    task = send_report.delay(req.user_id, req.email)
    return {"task_id": task.id, "status": "queued"}

Running the Worker

The FastAPI server only produces messages. You must start a separate worker process to consume and execute them.

  • celery_app after -A points to the module and instance.
  • --loglevel=info shows each task as it is received and succeeds.
  • --concurrency=4 controls how many tasks run in parallel.

On Windows or macOS forking issues, add --pool=solo for development.

# terminal 1: API
uvicorn main:app --reload

# terminal 2: worker
celery -A celery_app.celery_app worker --loglevel=info --concurrency=4

Polling Task Status with the Result Backend

Because we configured a result backend, we can look up a job's state and return value later using its id.

Build an AsyncResult from the id, bound to the same Celery app. Useful states:

  • PENDING — unknown/not started.
  • STARTED — picked up by a worker.
  • SUCCESS — finished; .result holds the return value.
  • FAILURE — raised an exception.

Clients poll this status endpoint until the task is ready.

# main.py (continued)
from celery.result import AsyncResult
from celery_app import celery_app

@app.get("/reports/{task_id}")
def get_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    payload = {"task_id": task_id, "state": result.state}
    if result.successful():
        payload["result"] = result.result
    return payload

apply_async: Countdown, ETA and Retries

.delay(*args) is shorthand for .apply_async(args=...). The longer form unlocks scheduling and routing options:

  • countdown=10 — wait 10 seconds before executing.
  • eta=datetime(...) — run at a specific time.
  • queue="emails" — route to a named queue.
  • retry=True with retry_policy — retry on broker errors.
from tasks import send_report

send_report.apply_async(
    args=[42, "user@example.com"],
    countdown=10,
    queue="reports",
)

Retrying Failed Tasks

Transient failures (a flaky API, a timeout) should be retried, not lost. Bind the task with bind=True so self is available, then call self.retry().

  • max_retries caps the attempts.
  • default_retry_delay or countdown backs off between tries.
  • autoretry_for can retry automatically for specific exceptions.

After exhausting retries, the task ends in the FAILURE state.

from celery_app import celery_app

@celery_app.task(
    bind=True,
    name="tasks.charge_card",
    max_retries=3,
    autoretry_for=(ConnectionError,),
    retry_backoff=True,
)
def charge_card(self, order_id: int):
    try:
        process_payment(order_id)
    except ConnectionError as exc:
        raise self.retry(exc=exc, countdown=5)

FastAPI BackgroundTasks vs. Celery

FastAPI ships a lightweight BackgroundTasks helper. It runs work in the same process, after the response is sent. Know when each fits:

  • BackgroundTasks — quick, fire-and-forget jobs (send one email, write a log). No retries, no result tracking, lost if the process restarts.
  • Celery — heavy, long, retryable, or schedulable jobs that need durability, horizontal scaling across machines, and result/state visibility.

Rule of thumb: if losing the job on a crash is unacceptable, or the work is CPU/time heavy, reach for Celery.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("audit.log", "a") as f:
        f.write(message + "\n")

@app.post("/click")
def click(bt: BackgroundTasks):
    bt.add_task(write_log, "user clicked")
    return {"ok": True}

A Standalone Queue Simulation

You cannot run a real broker inside an online judge, but the producer/consumer pattern behind Celery is simple. This pure-Python example mirrors the idea: jobs are enqueued, a worker drains the queue, and results are collected by id — exactly the flow Celery automates over Redis.

from collections import deque

queue = deque()
results = {}

def enqueue(task_id, user_id, email):
    queue.append((task_id, user_id, email))
    results[task_id] = "PENDING"

def worker():
    while queue:
        task_id, user_id, email = queue.popleft()
        results[task_id] = {
            "user_id": user_id,
            "sent_to": email,
            "status": "SUCCESS",
        }

enqueue("t1", 42, "a@x.com")
enqueue("t2", 7, "b@x.com")
worker()

for tid in ("t1", "t2"):
    print(tid, results[tid])

Quick Check

Test your understanding of how a FastAPI endpoint should hand off work to Celery.

Recap

You wired Celery into a FastAPI app end to end:

  • Created one Celery instance with a Redis broker and a result backend.
  • Defined JSON-serializable tasks with @celery_app.task and explicit names.
  • Dispatched jobs from endpoints with .delay(), returning a task id and 202 Accepted instead of blocking.
  • Ran a separate celery ... worker process to consume the queue.
  • Polled job state and results via AsyncResult.
  • Used apply_async for countdowns/queues and self.retry() for resilient retries.
  • Chose between FastAPI BackgroundTasks (light, in-process) and Celery (durable, scalable, retryable).

The golden rule: endpoints enqueue and return fast; workers do the heavy lifting.

Frequently asked questions

Is the “Wiring Celery Workers to a FastAPI App” lesson free?

Yes — the full text of “Wiring Celery Workers to a FastAPI App” is free to read here on the web, and the FastAPI Backend Development Bootcamp course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the FastAPI Backend Development Bootcamp course, upgrade to CoddyKit PRO.

What will I learn in “Wiring Celery Workers to a FastAPI App”?

Configure Celery with a Redis broker, define tasks, and dispatch jobs from endpoints with result backends. You practise FastAPI Backend Development Bootcamp with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start FastAPI Backend Development Bootcamp?

No prior experience is required. FastAPI Backend Development Bootcamp on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Wiring Celery Workers to a FastAPI App” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this FastAPI Backend Development Bootcamp lesson?

Yes. Every FastAPI Backend Development Bootcamp lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Lightweight Offloading with BackgroundTasks
  2. Wiring Celery Workers to a FastAPI App
  3. Retries, Idempotency and Dead-Letter Handling
  4. Scheduled and Periodic Jobs with Celery Beat
← Back to FastAPI Backend Development Bootcamp