ربط Celery Workers بتطبيق FastAPI
تهيئة Celery باستخدام وسيط Redis، وتعريف المهام، وإرسال الوظائف من نقاط النهاية مع مخازن نتائج.
ربط Celery Workers بتطبيق FastAPI درس مجاني في FastAPI Backend Development Bootcamp على CoddyKit. هذا هو الدرس 2 من أصل 4. يمكنك قراءة الدرس كاملاً أدناه مجاناً — ثم تمرن عليه مباشرة في المتصفح باستخدام محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7. هذا الدرس جزء من مسار التعلم في FastAPI Backend Development Bootcamp، وتقدمك يتزامن عبر الويب وتطبيق CoddyKit. تتضمن دورة FastAPI Backend Development Bootcamp 4 دروس في المجموع.
بعض أجزاء هذا الدرس لم تُترجم بعد وتظهر باللغة الإنجليزية.
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 theredisclient.- 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:7Creating 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
nameso 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_appafter-Apoints to the module and instance.--loglevel=infoshows each task as it is received and succeeds.--concurrency=4controls 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=4Polling 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;.resultholds 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 payloadapply_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=Truewithretry_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_retriescaps the attempts.default_retry_delayorcountdownbacks off between tries.autoretry_forcan 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
Celeryinstance with a Redis broker and a result backend. - Defined JSON-serializable tasks with
@celery_app.taskand explicit names. - Dispatched jobs from endpoints with
.delay(), returning a task id and 202 Accepted instead of blocking. - Ran a separate
celery ... workerprocess to consume the queue. - Polled job state and results via
AsyncResult. - Used
apply_asyncfor countdowns/queues andself.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.
تعلم FastAPI Backend Development Bootcamp مع معلم ذكاء اصطناعي — مجانًا
اكتب وقم بتشغيل أكوادك الفعلية في المتصفح، واحصل على مساعدة فورية من معلم ذكاء اصطناعي متاح 24/7، واستمر من حيث توقفت على الويب أو في التطبيق.
- الدورات
- 21
- الدروس
- 84
الأسئلة الشائعة
هل درس «ربط Celery Workers بتطبيق FastAPI» مجاني؟
نعم — نص درس «ربط Celery Workers بتطبيق FastAPI» كامل متاح مجاناً هنا على الويب. لتمرينه بشكل تفاعلي (محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7) وفتح باقي دورة FastAPI Backend Development Bootcamp، انتقل إلى CoddyKit PRO. تتضمن دورة FastAPI Backend Development Bootcamp 4 دروس في المجموع.
ماذا ستتعلم في «ربط Celery Workers بتطبيق FastAPI»؟
تهيئة Celery باستخدام وسيط Redis، وتعريف المهام، وإرسال الوظائف من نقاط النهاية مع مخازن نتائج. تتمرن على FastAPI Backend Development Bootcamp مع أكواد عملية تشغلها مباشرة في المتصفح، ومدرس ذكاء اصطناعي متاح 24/7 يجيب على أسئلتك أثناء عملك.
هل أحتاج إلى خبرة سابقة لأبدأ FastAPI Backend Development Bootcamp؟
لا تُشترط خبرة سابقة. FastAPI Backend Development Bootcamp على CoddyKit منظم للمبتدئين حتى المتقدمين، لذا يمكنك البدء من هنا أو من البداية والتقدم بسرعتك الخاصة. هذا هو الدرس 2 من أصل 4.
كم من الوقت يستغرق درس «ربط Celery Workers بتطبيق FastAPI»؟
معظم دروس CoddyKit تستغرق حوالي 5–10 دقائق. كل منها موجز وتفاعلي، لذا تحرز تقدماً مستمراً وتستأنف من حيث توقفت عبر الويب والتطبيق.
هل يمكنني كتابة وتشغيل أكواد في درس FastAPI Backend Development Bootcamp هذا؟
نعم. كل درس في FastAPI Backend Development Bootcamp يتضمن محرر أكواد مدمج، لذا تكتب وتشغل أكواداً حقيقية مباشرة في متصفحك وتحصل على تعليقات فورية من الذكاء الاصطناعي — بدون إعداد محلي.
جميع الدروس في هذه الدورة
- تفريغ المهام الخفيف باستخدام BackgroundTasks
- ربط Celery Workers بتطبيق FastAPI
- إعادة المحاولة وقابلية التكرار ومعالجة الرسائل الميتة
- المهام المجدولة والدورية باستخدام Celery Beat