Background Tasks and Celery — Asynchronous Task Queues
BackgroundTasks is like a restaurant call-number system—you get a number (API response) immediately after ordering, and you're notified when your food is ready; Celery is like a central kitchen—multiple chefs prepare different orders simultaneously, and each order's status can be tracked, with retries in case of failure.
1. What You'll Learn
- FastAPI Built-in
BackgroundTasks: A Quick Solution for Simple Scenarios - Celery Architecture: Worker / Broker (Redis) / Backend / Flower Monitoring
- Integrating Celery with FastAPI: Task Definitions, Triggers, and Status Query Endpoints
- Task Retry and Error Handling:
@app.task(retry=3, acks_late=True) - Alice Scenario: Batch Price Scraping—The API immediately returns a task ID, and the Celery Worker asynchronously processes the collection of prices for millions of products
2. Alice's True Story
(1) Pain Point: Time-consuming tasks block API responses
Alice's PriceTracker needs to crawl the prices of millions of products in bulk, and a single crawl takes 30 minutes. If processed synchronously, the API requests would time out, and Bob's frontend would have to wait 30 minutes to receive a response—resulting in a terrible user experience. Meanwhile, FastAPI's BackgroundTasks can only run within the current process, and restarting a worker would cause the task to be lost.
(2) Solution for Celery Distributed Queues
Celery places time-consuming tasks into a message queue (Redis Broker), where Worker processes consume them asynchronously, and the API immediately returns the task ID. Tasks are automatically retried if they fail, Workers can be horizontally scaled, and process restarts do not affect tasks in the queue.
PYTHON
from celery import Celery
celery_app = Celery("pricetracker", broker="redis://localhost:6379/0")
@celery_app.task(bind=True, max_retries=3)
def scrape_prices(self, product_ids: list[int]):
# Async price scraping - runs in Celery Worker
...
(3) Revenue
Scraping a million products has gone from "blocking the API for 30 minutes" to "returning a task ID in 1 second + background processing." The number of workers can be scaled from 1 to 10, reducing scraping time from 30 minutes to 3 minutes. Tasks automatically retry three times upon failure, increasing the success rate from 95% to 99.9%.
3. Lightweight Solutions for Background Tasks
(1) Applicable Scenarios
(1) ▶ Example: Sending Notifications with BackgroundTasks
PYTHON
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
app = FastAPI()
class PriceAlertRequest(BaseModel):
product_id: int
target_price: float
email: str
def send_price_alert_email(email: str, product_id: int, price: float):
# Simulate email sending (do NOT use await here)
print(f"Sending alert to {email}: Product {product_id} hit ${price}")
@app.post("/alerts")
async def create_alert(alert: PriceAlertRequest, bg: BackgroundTasks):
# Add task to run after response is sent
bg.add_task(send_price_alert_email, alert.email, alert.product_id, alert.target_price)
return {"message": "Alert created", "product_id": alert.product_id}
Output:
TEXT
# Function defined successfully
(2) Decision Tree: BackgroundTasks vs. Celery
flowchart TD
Start{Need background task?} --> Time{Takes > 1 min?}
Time -->|No| Simple[Use BackgroundTasks]
Time -->|Yes| Retry{Need retry/resilience?}
Retry -->|No| Simple
Retry -->|Yes| Scale{Need horizontal scaling?}
Scale -->|No| Simple
Scale -->|Yes| Celery[Use Celery]
Simple -->|Pros| P1[Simple, no infra]
Simple -->|Cons| C1[No retry, no scale, lost on restart]
Celery -->|Pros| P2[Retry, scale, persistent, monitor]
Celery -->|Cons| C2[Redis + Worker infrastructure]
| Dimension | Background Tasks | Celery |
|---|---|---|
| Complexity | Zero configuration | Requires Redis + Worker |
| Persistence | Process Memory | Redis Persistence |
| Retry | None | Built-in retry mechanism |
| Scalability | Single-process | Worker-based horizontal scaling |
| Monitoring | None | Flower Dashboard |
| Use Case | Lightweight tasks < 1 second | Time-consuming tasks > 1 minute |
4. Celery Architecture
(1) Overview of the Architecture
flowchart TD
API[FastAPI App] -->|Enqueue Task| Broker[(Redis Broker)]
Broker -->|Consume Task| Worker1[Celery Worker 1]
Broker -->|Consume Task| Worker2[Celery Worker 2]
Broker -->|Consume Task| WorkerN[Celery Worker N]
Worker1 -->|Store Result| Backend[(Redis Backend)]
Worker2 -->|Store Result| Backend
WorkerN -->|Store Result| Backend
API -->|Query Status| Backend
Flower[Flower Monitor] -->|Observe| Broker
Flower -->|Observe| Backend
| Component | Function | Recommendation |
|---|---|---|
| Broker | Task Message Queue | Redis |
| Backend | Result Storage | Redis |
| Worker | Task Execution Process | celery -A app worker |
| Flower | Monitoring Dashboard | celery -A app flower |
(1) ▶ Example: Celery Configuration
PYTHON
# app/core/celery_app.py
from celery import Celery
celery_app = Celery(
"pricetracker",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/1",
)
celery_app.conf.update(
task_serializer="json",
accept_content=["json"],
result_serializer="json",
timezone="UTC",
enable_utc=True,
task_track_started=True,
task_acks_late=True, # Ack after execution, not before
worker_prefetch_multiplier=4,
result_expires=3600, # Results expire after 1 hour
)
Output:
TEXT
# Execution Successful
5. Celery Task Definition and Triggering
(1) Task Definition
(1) ▶ Example: Price Scraping Task
PYTHON
# app/tasks/price_scraping.py
from app.core.celery_app import celery_app
import asyncio
from sqlalchemy import select
@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def scrape_product_prices(self, product_ids: list[int]):
"""Scrape current prices for given products."""
try:
for pid in product_ids:
# Simulate scraping (in production: HTTP requests to price sources)
price = fetch_price_from_source(pid)
# Store in database
save_price_record(pid, price)
return {"scraped": len(product_ids), "status": "success"}
except Exception as exc:
# Retry with exponential backoff
raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))
@celery_app.task(bind=True)
def bulk_scrape_all(self, total_products: int = 1000000, batch_size: int = 1000):
"""Scrape all products in batches - chord pattern."""
batches = [
list(range(i, min(i + batch_size, total_products)))
for i in range(0, total_products, batch_size)
]
# Fan out to individual scrape tasks
for batch in batches:
scrape_product_prices.delay(batch)
return {"total_batches": len(batches), "status": "started"}
Output:
TEXT
# Function defined successfully
(2) Task Status Transitions
stateDiagram-v2
[*] --> PENDING: Task created
PENDING --> STARTED: Worker picks up
STARTED --> PROGRESS: Running (optional)
PROGRESS --> SUCCESS: Completed
PROGRESS --> FAILURE: Error occurred
STARTED --> FAILURE: Error occurred
FAILURE --> RETRY: max_retries not reached
RETRY --> PENDING: Re-queued
FAILURE --> [*]: max_retries exceeded
SUCCESS --> [*]
(2) ▶ Example: FastAPI Endpoint Triggers a Celery Task
PYTHON
from fastapi import FastAPI, Depends
from app.core.celery_app import celery_app
from app.tasks.price_scraping import scrape_product_prices, bulk_scrape_all
app = FastAPI()
@app.post("/api/v1/scrape/prices")
async def trigger_scrape(
product_ids: list[int],
user=Depends(require_subscription("pro")),
):
# Trigger Celery task - returns task ID immediately
task = scrape_product_prices.delay(product_ids)
return {"task_id": task.id, "status": "pending"}
@app.post("/api/v1/scrape/bulk")
async def trigger_bulk_scrape(
user=Depends(require_subscription("enterprise")),
):
task = bulk_scrape_all.delay(total_products=1000000)
return {"task_id": task.id, "status": "pending"}
Output:
TEXT
# Function defined successfully
(3) ▶ Example: Task Status Query Endpoint
PYTHON
from celery.result import AsyncResult
@app.get("/api/v1/tasks/{task_id}")
async def get_task_status(task_id: str):
result = AsyncResult(task_id, app=celery_app)
response = {
"task_id": task_id,
"status": result.status,
}
if result.ready():
if result.successful():
response["result"] = result.result
else:
response["error"] = str(result.result)
elif result.state == "PROGRESS":
response["progress"] = result.info
return response
Output:
TEXT
# Function defined successfully
6. Running and Monitoring Celery Workers
(1) ▶ Example: Starting Worker and Flower
BASH
# Start Celery Worker
celery -A app.core.celery_app worker --loglevel=info --concurrency=4
# Start Flower monitoring dashboard
celery -A app.core.celery_app flower --port=5555
# Visit http://localhost:5555 for monitoring dashboard
Output:
TEXT
# Command executed successfully
(2) ▶ Example: Task Progress Report
PYTHON
from celery import current_task
@celery_app.task(bind=True)
def scrape_with_progress(self, product_ids: list[int]):
total = len(product_ids)
for i, pid in enumerate(product_ids):
# Process each product
price = fetch_price_from_source(pid)
save_price_record(pid, price)
# Report progress
self.update_state(
state="PROGRESS",
meta={"current": i + 1, "total": total, "percent": (i + 1) / total * 100},
)
return {"scraped": total}
Output:
TEXT
# Function defined successfully
❓ FAQ
Q When are BackgroundTasks executed?
A They are executed after the response is sent to the client. If a task throws an exception, it does not affect the response that has already been sent, but it will be logged.
Q Should Celery Worker and FastAPI run in the same process?
A No. Worker is a separate process that is deployed and scaled independently. FastAPI is only responsible for triggering tasks, while Worker is responsible for executing them.
Q What is the difference between using Redis as a Broker and as a Backend?
A The Broker is a task queue (tasks awaiting execution), while the Backend is a result store (results of completed tasks). You can use different databases within the same Redis instance (e.g., DB 0 and DB 1).
Q What is the purpose of
task_acks_late=True?A By default, a worker acknowledges a task as soon as it receives it; if it crashes during execution, the task is lost. Setting
acks_late=True causes the worker to acknowledge the task only after execution is complete; if it crashes, the task will be re-executed by another worker.Q Can I use
async/await in Celery tasks?A Celery tasks are synchronous functions. To call asynchronous code, wrap it in
asyncio.run(). Alternatively, use the eventlet or gevent concurrency models in the Worker.Q How do you handle progress tracking for tasks involving millions of items?
A Use the "chord" pattern: Break the task down into 1,000 subtasks (each processing 1,000 items), and once the subtasks are complete, the "chord" callback aggregates the results. The front end polls the status of the "chord" tasks.
📖 Summary
- BackgroundTasks are suitable for lightweight tasks that take less than 1 second (sending emails, writing logs); they require no configuration but do not support retries or persistence.
- Celery is a production-grade task queue: Broker (Redis queue) + Worker (execution) + Backend (result storage)
@app.task(bind=True, max_retries=3)Define retryable tasks;self.retry()Trigger a retry- FastAPI triggers a task via
.delay()and queries the status viaAsyncResult; the API immediately returns the task ID - Flower provides a visual monitoring dashboard that lets you view worker status, task progress, and success rates in real time
📝 Exercises
- Basic Problem (Difficulty ⭐): Use
BackgroundTasksto implement an endpoint that sends a notification email in the background (simulating log output) after a product is created, and the API immediately returns the creation result. Hint:bg: BackgroundTasks+bg.add_task(fn, args) - Advanced Exercise (Difficulty ⭐⭐): Configure Celery (Redis Broker + Backend), define the
scrape_pricestask (max_retries=3), have a FastAPI endpoint trigger the task and return the task_id, and have another endpoint query the task status. Hint:celery_app.delay()+AsyncResult(task_id) - Challenge (Difficulty: ⭐⭐⭐): Implement a batch scraping task with progress reporting—
scrape_with_progressreports the progress percentage toself.update_state(state="PROGRESS"), the front end polls/tasks/{task_id}to display a progress bar, and Enterprise users can trigger scraping at the million-level. Hint: Useself.update_state()+result.infoto retrieve progress
---|
