PricingLoginStart free
FastAPI Backend Development Bootcamp · Ders

Celery Çalışanlarını FastAPI Uygulamasına Bağlama

Celery'yi Redis aracısıyla yapılandırın, görevleri tanımlayın ve sonuç arka uçlarıyla uç noktalardan işleri gönderin.

2. ders / 413 adım

Celery Çalışanlarını FastAPI Uygulamasına Bağlama, CoddyKit'te ücretsiz bir FastAPI Backend Development Bootcamp dersidir. Bu, 4 dersinin 2. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, FastAPI Backend Development Bootcamp öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. FastAPI Backend Development Bootcamp kursu toplamda 4 dersten oluşur.

Bu dersin bazı bölümleri henüz çevrilmemiş olup İngilizce olarak gösterilmektedir.

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.

İnteraktif derste cevapla

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.

Başlamak ücretsiz

Yapay zeka eğitmeniyle FastAPI Backend Development Bootcamp öğren — ücretsiz

Tarayıcında gerçek kod yaz ve çalıştır, 7/24 yapay zeka eğitmeninden anında yardım al; web'de ya da uygulamada kaldığın yerden devam et.

Kurslar
21
Dersler
84
Ücretsiz öğrenmeye başlaTüm kursu görüntüle

Sıkça Sorulan Sorular

“Celery Çalışanlarını FastAPI Uygulamasına Bağlama” dersi ücretsiz mi?

Evet — “Celery Çalışanlarını FastAPI Uygulamasına Bağlama” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve FastAPI Backend Development Bootcamp kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. FastAPI Backend Development Bootcamp kursu toplamda 4 dersten oluşur.

“Celery Çalışanlarını FastAPI Uygulamasına Bağlama” dersinde ne öğreneceğim?

Celery'yi Redis aracısıyla yapılandırın, görevleri tanımlayın ve sonuç arka uçlarıyla uç noktalardan işleri gönderin. FastAPI Backend Development Bootcamp ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.

FastAPI Backend Development Bootcamp öğrenmeye başlamak için deneyim gerekli mi?

Önceden deneyim gerekmez. CoddyKit'te FastAPI Backend Development Bootcamp, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 2. dersidir.

“Celery Çalışanlarını FastAPI Uygulamasına Bağlama” dersi ne kadar sürer?

Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.

Bu FastAPI Backend Development Bootcamp dersinde kod yazıp çalıştırabilir miyim?

Evet. Her FastAPI Backend Development Bootcamp dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.

Bu kursun tüm dersleri

  1. BackgroundTasks ile Hafif İş Yükü Aktarımı
  2. Celery Çalışanlarını FastAPI Uygulamasına Bağlama
  3. Yeniden Denemeler, İdempotensi ve Teslim Edilemeyen Mesaj Yönetimi
  4. Celery Beat ile Zamanlanmış ve Periyodik İşler
← FastAPI Backend Development Bootcamp Sayfasına Dön