0Pricing
FastAPI Backend Development Bootcamp · Lektion

Celery-Worker mit einer FastAPI-App verbinden

Konfigurieren Sie Celery mit einem Redis-Broker, definieren Sie Tasks und starten Sie Jobs aus Endpunkten mit Result-Backends.

Celery-Worker mit einer FastAPI-App verbinden ist eine kostenlose FastAPI Backend Development Bootcamp-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des FastAPI Backend Development Bootcamp-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der FastAPI Backend Development Bootcamp-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

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.

Häufig gestellte Fragen

Ist die Lektion „Celery-Worker mit einer FastAPI-App verbinden“ kostenlos?

Ja — der vollständige Text von „Celery-Worker mit einer FastAPI-App verbinden“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des FastAPI Backend Development Bootcamp-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der FastAPI Backend Development Bootcamp-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Celery-Worker mit einer FastAPI-App verbinden“?

Konfigurieren Sie Celery mit einem Redis-Broker, definieren Sie Tasks und starten Sie Jobs aus Endpunkten mit Result-Backends. Du übst FastAPI Backend Development Bootcamp mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um FastAPI Backend Development Bootcamp zu starten?

Keine Vorkenntnisse erforderlich. FastAPI Backend Development Bootcamp auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „Celery-Worker mit einer FastAPI-App verbinden“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser FastAPI Backend Development Bootcamp-Lektion Code schreiben und ausführen?

Ja. Jede FastAPI Backend Development Bootcamp-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Leichtgewichtige Auslagerung mit BackgroundTasks
  2. Celery-Worker mit einer FastAPI-App verbinden
  3. Retries, Idempotenz und Dead-Letter-Verarbeitung
  4. Geplante und periodische Jobs mit Celery Beat
← Zurück zu FastAPI Backend Development Bootcamp