Celery-workers koppelen aan een FastAPI-app
Configureer Celery met een Redis-broker, definieer taken en verstuur jobs vanuit endpoints met result backends.
Celery-workers koppelen aan een FastAPI-app is een gratis Bootcamp backendontwikkeling met FastAPI-les op CoddyKit. Dit is les 2 van 4. Je kunt 3 lessen uit dit leerpad gratis volledig lezen — daarna ontgrendelt CoddyKit PRO alle lessen, plus praktische oefeningen met een ingebouwde code-editor en een AI-tutor die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject Bootcamp backendontwikkeling met FastAPI. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus Bootcamp backendontwikkeling met FastAPI bevat in totaal 4 lessen.
Waarom Celery in een FastAPI-app?
Requestafhandelaars van FastAPI moeten snel terugkeren. Werk zoals e-mails versturen, rapporten genereren, afbeeldingen schalen of trage API's van derden aanroepen kan seconden duren, waardoor je workerproces wordt geblokkeerd en de verwerkingscapaciteit afneemt.
Celery is een gedistribueerde takenwachtrij. Je plaatst een taak in een wachtrij en een afzonderlijke pool van workerprocessen voert die buiten de request/responsecyclus uit.
- Broker — het berichten transport dat taken in de wachtrij bevat (we gebruiken Redis).
- Worker — een langdurig actief proces dat taken ophaalt en uitvoert.
- Resultaatbackend — een optionele opslag voor retourwaarden en de status van taken.
Het endpoint blijft snel en antwoordt alleen: "geaccepteerd, dit is je taak-id."
De onderdelen installeren
Installeer Celery met de Redis-extra en zorg daarnaast voor een Redis-server die bereikbaar is vanaf zowel je API als je workers.
celery[redis]installeert Celery en deredis-client.- Start Redis lokaal met Docker:
docker run -p 6379:6379 redis.
Zowel het FastAPI-proces als het workerproces importeert hetzelfde Celery-applicatieobject, dus ze moeten dezelfde codebase en broker-URL delen.
# requirements.txt
fastapi
uvicorn[standard]
celery[redis]
redis
# install
# pip install -r requirements.txt
# run redis: docker run -p 6379:6379 redis:7De Celery-app maken
Definieer één Celery-applicatie-instantie in een eigen module (vaak worker.py of celery_app.py). Deze heeft een naam, een broker-URL en een URL voor de resultaatbackend nodig.
broker— waar taken in de wachtrij worden geplaatst (Redis-database 0).backend— waar resultaten en statussen worden opgeslagen (Redis-database 1, voor de duidelijkheid gescheiden).
Het eerste argument ("worker") wordt de standaardprefix voor taaknamen.
# 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,
)Je eerste taak definiëren
Een taak is gewoon een functie met de decorator @celery_app.task. Wanneer je deze normaal aanroept, wordt ze inline uitgevoerd; wanneer je haar aanroept met .delay() of .apply_async(), wordt ze geserialiseerd en naar de broker gestuurd.
- Argumenten moeten JSON-serialiseerbaar zijn (gebruik id's en primitieve waarden, geen ORM-objecten).
- Geef de taak een expliciete
name, zodat het later hernoemen van de functie berichten in de wachtrij niet ongeldig maakt.
# 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"}Een taak vanuit een endpoint verzenden
Roep binnen een FastAPI-route .delay(...) aan om de taak in de wachtrij te plaatsen. Dit retourneert onmiddellijk een AsyncResult — er wordt niet gewacht tot het werk klaar is.
Antwoord met task.id en HTTP 202 Accepted om aan te geven dat de aanvraag voor verwerking is geaccepteerd, maar nog niet is voltooid.
Roep nooit .get() aan in de requestafhandelaar — daarmee blokkeer je de eventlus totdat de taak klaar is en doe je het hele doel teniet.
# 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"}De worker uitvoeren
De FastAPI-server produceert alleen berichten. Je moet een afzonderlijk workerproces starten om ze te consumeren en uit te voeren.
celery_appna-Averwijst naar de module en de instantie.--loglevel=infotoont elke taak wanneer deze wordt ontvangen en succesvol wordt uitgevoerd.--concurrency=4bepaalt hoeveel taken parallel worden uitgevoerd.
Voeg bij problemen met forking op Windows of macOS --pool=solo toe voor ontwikkeling.
# terminal 1: API
uvicorn main:app --reload
# terminal 2: worker
celery -A celery_app.celery_app worker --loglevel=info --concurrency=4De taakstatus opvragen met de resultaatbackend
Omdat we een resultaatbackend hebben ingesteld, kunnen we de status en retourwaarde van een taak later opzoeken met de id.
Maak op basis van de id een AsyncResult die aan dezelfde Celery-app is gekoppeld. Handige statussen:
PENDING— onbekend/nog niet gestart.STARTED— door een worker opgepakt.SUCCESS— voltooid;.resultbevat de retourwaarde.FAILURE— er is een uitzondering opgetreden.
Clients vragen dit statusendpoint steeds opnieuw op totdat de taak klaar is.
# 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 payloadapply_async: aftellen, ETA en opnieuw proberen
.delay(*args) is een verkorte vorm van .apply_async(args=...). De langere vorm biedt opties voor planning en routering:
countdown=10— wacht 10 seconden voordat de taak wordt uitgevoerd.eta=datetime(...)— voer de taak op een specifiek tijdstip uit.queue="emails"— routeer de taak naar een wachtrij met een naam.retry=Truemetretry_policy— probeer opnieuw bij brokerfouten.
from tasks import send_report
send_report.apply_async(
args=[42, "user@example.com"],
countdown=10,
queue="reports",
)Mislukte taken opnieuw proberen
Tijdelijke fouten (een onbetrouwbare API, een time-out) moet je opnieuw proberen, niet negeren. Bind de taak met bind=True zodat self beschikbaar is en roep daarna self.retry() aan.
max_retriesbegrenst het aantal pogingen.default_retry_delayofcountdownzorgt voor wachttijd tussen pogingen.autoretry_forkan automatisch opnieuw proberen voor specifieke uitzonderingen.
Na het opgebruiken van alle pogingen eindigt de taak met de status FAILURE.
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 versus Celery
FastAPI wordt geleverd met een lichtgewicht helper BackgroundTasks. Deze voert werk uit in hetzelfde proces, nadat de respons is verzonden. Weet wanneer je welke optie gebruikt:
- BackgroundTasks — snelle taken waarbij je niet op een resultaat wacht (één e-mail versturen, een logregel schrijven). Geen nieuwe pogingen, geen resultaatregistratie en verlies bij een herstart van het proces.
- Celery — zware, langdurige, opnieuw uitvoerbare of planbare taken waarvoor duurzaamheid, horizontaal schalen over machines en inzicht in resultaten en status nodig zijn.
Vuistregel: als het onacceptabel is dat een taak bij een crash verloren gaat, of als het werk veel CPU of tijd kost, kies je 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}Een zelfstandige simulatie van een wachtrij
Je kunt geen echte broker uitvoeren in een online beoordelingsomgeving, maar het producent-consumentpatroon achter Celery is eenvoudig. Dit pure Python-voorbeeld weerspiegelt het idee: taken worden in de wachtrij geplaatst, een worker maakt de wachtrij leeg en resultaten worden op id verzameld — precies de stroom die Celery via Redis automatiseert.
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])Korte controle
Test je begrip van hoe een FastAPI-endpoint werk aan Celery moet overdragen.
Samenvatting
Je hebt Celery van begin tot eind in een FastAPI-app geïntegreerd:
- Eén
Celery-instantie gemaakt met een Redis-broker en een resultaatbackend. - JSON-serialiseerbare taken met
@celery_app.tasken expliciete namen gedefinieerd. - Taken vanuit endpoints verzonden met
.delay()en een taak-id en 202 Accepted teruggegeven zonder te blokkeren. - Een afzonderlijk
celery ... worker-proces uitgevoerd om de wachtrij te consumeren. - De status en resultaten van taken opgevraagd via
AsyncResult. apply_asyncgebruikt voor aftellen en wachtrijen enself.retry()voor robuuste nieuwe pogingen.- Gekozen tussen FastAPI
BackgroundTasks(licht, binnen hetzelfde proces) en Celery (duurzaam, schaalbaar, opnieuw uitvoerbaar).
De gouden regel: endpoints plaatsen taken in de wachtrij en keren snel terug; workers doen het zware werk.
Leer Bootcamp backendontwikkeling met FastAPI met een AI-tutor — gratis
Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.
- Cursussen
- 21
- Lessen
- 84
Veelgestelde vragen
Is de les “Celery-workers koppelen aan een FastAPI-app” gratis?
Ja — je kunt hier op het web alle 3 lessen van het leerpad Bootcamp backendontwikkeling met FastAPI, waaronder “Celery-workers koppelen aan een FastAPI-app”, gratis volledig lezen. Daarna ontgrendelt CoddyKit PRO alle lessen, plus interactieve oefeningen met een ingebouwde code-editor en een AI-tutor die 24/7 beschikbaar is. De cursus Bootcamp backendontwikkeling met FastAPI bevat in totaal 4 lessen.
Wat leer ik in “Celery-workers koppelen aan een FastAPI-app”?
Configureer Celery met een Redis-broker, definieer taken en verstuur jobs vanuit endpoints met result backends. Je oefent met Bootcamp backendontwikkeling met FastAPI door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.
Heb ik ervaring nodig om met Bootcamp backendontwikkeling met FastAPI te beginnen?
Ervaring vooraf is niet nodig. Bootcamp backendontwikkeling met FastAPI op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 2 van 4.
Hoe lang duurt de les “Celery-workers koppelen aan een FastAPI-app”?
De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.
Kan ik code schrijven en uitvoeren in deze les over Bootcamp backendontwikkeling met FastAPI?
Ja. Elke les over Bootcamp backendontwikkeling met FastAPI bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.
Alle lessen in deze cursus
- Lichtgewicht offloading met BackgroundTasks
- Celery-workers koppelen aan een FastAPI-app
- Retries, idempotentie en dead-letterafhandeling
- Geplande en periodieke jobs met Celery Beat