PricingLoginStart free
FastAPI Backend Development Bootcamp · Lekcja

Podłączanie workerów Celery do aplikacji FastAPI

Konfiguruj Celery z brokerem Redis, definiuj zadania i wysyłaj je z endpointów za pomocą backendów wyników.

Lekcja 2 z 413 kroki

Podłączanie workerów Celery do aplikacji FastAPI to bezpłatna lekcja FastAPI Backend Development Bootcamp na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej FastAPI Backend Development Bootcamp, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.

Części tej lekcji nie zostały jeszcze przetłumaczone i są wyświetlane po angielsku.

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.

Odpowiedz w interaktywnej lekcji

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.

Bezpłatny start

Ucz się FastAPI Backend Development Bootcamp dzięki korepetycjom AI — za darmo

Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.

Kursy
21
Lekcje
84
Zacznij naukę bezpłatnieWyświetl pełny kurs

Często zadawane pytania

Czy lekcja „Podłączanie workerów Celery do aplikacji FastAPI” jest bezpłatna?

Tak — pełny tekst „Podłączanie workerów Celery do aplikacji FastAPI” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu FastAPI Backend Development Bootcamp, przejdź na CoddyKit PRO. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.

Co nauczysz się w „Podłączanie workerów Celery do aplikacji FastAPI”?

Konfiguruj Celery z brokerem Redis, definiuj zadania i wysyłaj je z endpointów za pomocą backendów wyników. Ćwiczysz FastAPI Backend Development Bootcamp z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć FastAPI Backend Development Bootcamp?

Nie wymagamy żadnego doświadczenia. FastAPI Backend Development Bootcamp w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.

Ile czasu zajmuje lekcja „Podłączanie workerów Celery do aplikacji FastAPI”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji FastAPI Backend Development Bootcamp?

Tak. Każda lekcja FastAPI Backend Development Bootcamp zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Lekkie odciążanie za pomocą BackgroundTasks
  2. Podłączanie workerów Celery do aplikacji FastAPI
  3. Ponowienia, idempotencja i obsługa niedostarczonych komunikatów
  4. Zadania zaplanowane i cykliczne za pomocą Celery Beat
← Powrót do FastAPI Backend Development Bootcamp