PricingLoginStart free
FastAPI Backend Development Bootcamp · Pelajaran

Menghubungkan Pekerja Celery ke Aplikasi FastAPI

Konfigurasikan Celery dengan perantara Redis, definisikan tugas, dan kirim pekerjaan dari titik akhir dengan backend hasil.

Pelajaran 2 dari 413 langkah

Menghubungkan Pekerja Celery ke Aplikasi FastAPI adalah pelajaran FastAPI Backend Development Bootcamp gratis di CoddyKit. Ini adalah pelajaran 2 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar FastAPI Backend Development Bootcamp, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus FastAPI Backend Development Bootcamp mencakup 4 pelajaran total.

Bagian dari pelajaran ini belum diterjemahkan dan ditampilkan dalam bahasa Inggris.

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.

Jawab di pelajaran interaktif

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.

Gratis untuk memulai

Belajar FastAPI Backend Development Bootcamp dengan tutor AI — gratis

Tulis dan jalankan kode asli di browser kamu, dapatkan bantuan instan dari tutor AI 24/7, dan lanjutkan di mana kamu tinggalkan di web atau aplikasi.

Kursus
21
Pelajaran
84
Mulai belajar gratisLihat kursus lengkapnya

Pertanyaan yang Sering Diajukan

Apakah pelajaran “Menghubungkan Pekerja Celery ke Aplikasi FastAPI” gratis?

Ya — teks lengkap “Menghubungkan Pekerja Celery ke Aplikasi FastAPI” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus FastAPI Backend Development Bootcamp, upgrade ke CoddyKit PRO. Kursus FastAPI Backend Development Bootcamp mencakup 4 pelajaran total.

Apa yang akan aku pelajari di “Menghubungkan Pekerja Celery ke Aplikasi FastAPI”?

Konfigurasikan Celery dengan perantara Redis, definisikan tugas, dan kirim pekerjaan dari titik akhir dengan backend hasil. Kamu berlatih FastAPI Backend Development Bootcamp dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.

Apakah aku perlu pengalaman untuk memulai FastAPI Backend Development Bootcamp?

Tidak diperlukan pengalaman sebelumnya. FastAPI Backend Development Bootcamp di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 2 dari 4.

Berapa lama pelajaran “Menghubungkan Pekerja Celery ke Aplikasi FastAPI” memakan waktu?

Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.

Bisakah aku menulis dan menjalankan kode dalam pelajaran FastAPI Backend Development Bootcamp ini?

Ya. Setiap pelajaran FastAPI Backend Development Bootcamp menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.

Semua pelajaran dalam kursus ini

  1. Pelimpahan Ringan dengan BackgroundTasks
  2. Menghubungkan Pekerja Celery ke Aplikasi FastAPI
  3. Percobaan Ulang, Idempotensi, dan Penanganan Surat Mati
  4. Pekerjaan Terjadwal dan Berkala dengan Celery Beat
← Kembali ke FastAPI Backend Development Bootcamp