Background Tasks & Workers

This lesson on Background Tasks & Workers is hands-on and example-driven. Offload slow, blocking I/O tasks like AI/LLM operations from FastAPI to Celery background workers backed by Redis. You will implement non-blocking job submission patterns, manage the end-to-end task lifecycle with AsyncResult, and deliver completed results using client polling and WebSockets.

What You'll Be Able To Do

• Configure a Celery application using isolated Redis database indices for broker and result storage.
• Define resilient background tasks with custom time limits, automatic retries, and bound exception handling.
• Implement FastAPI endpoints to asynchronously enqueue tasks, track job execution states, and revoke running jobs.
• Build client-side polling loops and persistent WebSocket connections to deliver job outputs without blocking request threads.
• Orchestrate local development startup sequences across Dockerized Redis, Celery workers, and FastAPI.

Detailed Concept Walkthrough

1. Async Job Queue Architecture

Synchronously processing slow operations starves web server worker threads and causes client timeouts. Offloading tasks to an asynchronous queue decouples request reception from background execution.

• Mechanism: When an HTTP request arrives, FastAPI immediately enqueues the job payload into a Redis message broker and returns a unique task receipt ID to the client without waiting for execution.
• Under the Hood: Celery worker processes run independently of the web server process, continuously listening to the Redis broker queue, pulling pending tasks, and executing them in isolated worker threads.
• Best Practice: Decouple the Redis broker from the Redis result backend by using distinct database numbers (e.g., DB 0 for queuing and DB 1 for results) to prevent state conflicts and memory contention under load.

from celery import Celery

# Configure Celery with isolated Redis databases for broker and backend
celery_app = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1"
)

Key Takeaway: FastAPI acts as the non-blocking dispatcher while Celery and Redis handle long-running compute out of band.

2. Celery Task Definition and Resilience

Celery tasks encapsulate background workloads with built-in fault tolerance, strict time bounds, and retry logic to survive external service failures.

• Mechanism: Tasks are registered using the @celery_app.task decorator with bind=True, giving the function access to its own runtime execution context via the self parameter.
• Under the Hood: Hard execution limits (time_limit) enforce worker safety by terminating tasks that hang indefinitely, preventing hung connections from permanently locking worker concurrency slots.
• Best Practice: Catch specific operational exceptions and call self.retry() with exponential backoff and a capped max_retries value to handle transient API or network disruptions.

import time
from celery_config import celery_app

@celery_app.task(bind=True, max_retries=3, time_limit=120)
def generate_llm_summary(self, prompt: str):
    try:
        # Simulate an external long-running AI API call
        time.sleep(10)
        return {"result": f"Processed prompt: {prompt}"}
    except Exception as exc:
        # Retry after 5 seconds if a transient failure occurs
        raise self.retry(exc=exc, countdown=5)

Key Takeaway: Bound tasks with configured timeouts and retries prevent runaway processes and transient network failures from draining worker capacity.

3. FastAPI Task Control Endpoints

FastAPI exposes non-blocking endpoints that map directly to Celery's task lifecycle: job submission, status inspection, and task revocation.

• Mechanism: The submission route invokes .delay(*args) on the Celery task, immediately returning a JSON payload containing the task ID and a status of PENDING.
• Under the Hood: The status endpoint initializes a Celery AsyncResult object with the given task ID to query Redis DB 1 for state changes (PENDING, SUCCESS, FAILURE) and extract the returned output upon completion.
• Execution Flow: If a client aborts a request, a dedicated cancellation endpoint triggers celery_app.control.revoke(task_id, terminate=True) to signal workers to immediately halt execution.

from fastapi import FastAPI, HTTPException
from celery.result import AsyncResult
from celery_config import celery_app
from tasks import generate_llm_summary

app = FastAPI()

@app.post("/tasks/summarize")
def enqueue_summary(prompt: str):
    task = generate_llm_summary.delay(prompt)
    return {"task_id": task.id, "status": "Queued"}

@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    return {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.ready() else None
    }

@app.post("/tasks/{task_id}/cancel")
def cancel_task(task_id: str):
    celery_app.control.revoke(task_id, terminate=True)
    return {"task_id": task_id, "status": "Revoked"}

Key Takeaway: Task control endpoints must remain entirely non-blocking by relying exclusively on task.delay and AsyncResult lookups.

4. Result Delivery: Polling vs. WebSockets

Clients retrieve async task outcomes either by periodically polling HTTP status endpoints or by establishing persistent WebSockets for instant push delivery.

• Mechanism: HTTP polling uses client-side intervals (e.g., every 2 seconds) to request /tasks/{id}, terminating the loop once the returned status equals SUCCESS or FAILURE.
• Under the Hood: WebSocket endpoints maintain an open full-duplex TCP connection, running an asynchronous server loop that checks AsyncResult.ready() and pushes the completed payload to the client immediately upon completion.
• Best Practice: Choose short-interval HTTP polling for simple, low-frequency workflows, and switch to WebSockets for low-latency, real-time UI updates to eliminate redundant HTTP request overhead.

import asyncio
from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/tasks/{task_id}")
async def task_websocket_status(websocket: WebSocket, task_id: str):
    await websocket.accept()
    try:
        while True:
            result = AsyncResult(task_id, app=celery_app)
            if result.ready():
                await websocket.send_json({"status": result.status, "result": result.result})
                break
            await websocket.send_json({"status": result.status})
            await asyncio.sleep(1)
    except WebSocketDisconnect:
        pass

Key Takeaway: WebSockets replace repetitive HTTP polling overhead with instant, event-driven result delivery.

Topics Covered in Background Tasks & Workers

1. Synchronous Bottleneck Problem (0:00 - 0:44) — Synchronous execution of slow operations ties up server threads and causes client request timeouts.
2. Async Job Queue Pattern (0:44 - 1:21) — FastAPI delegates long-running tasks to background queues and immediately returns receipt identifiers.
3. Submission and Polling Flow (1:21 - 2:21) — Clients submit jobs, workers store results, and clients poll the status endpoint until completion.
4. FastAPI, Celery, and Redis Roles (2:21 - 3:22) — FastAPI handles HTTP requests, Celery manages workers, and Redis serves as both broker and backend.
5. Environment Setup & Startup Order (3:22 - 3:49) — Services must be initialized sequentially starting with Redis, followed by Celery workers, then FastAPI.
6. Celery Task Configuration (3:49 - 4:42) — Celery tasks are configured with database isolation, execution time limits, and automatic retry handlers.
7. FastAPI Task Endpoints (4:42 - 5:18) — Endpoints are constructed to enqueue jobs with delay, inspect states via AsyncResult, and revoke tasks.
8. Client-Side HTTP Polling (5:18 - 5:47) — Client applications query the status endpoint at periodic intervals until the task state is complete.
9. WebSocket Result Push (5:47 - 6:13) — WebSockets replace client polling loops by streaming completed task results over persistent connections.

FastAPI Cheat Sheet

• celery -A tasks worker --loglevel=info — Starts a local Celery worker instance listening to the broker

  celery -A tasks worker --loglevel=info
  

• task.delay(*args, **kwargs) — Asynchronously enqueues a task and returns an AsyncResult receipt

  task = generate_llm_summary.delay(prompt="Summarize this text")
  

• AsyncResult(task_id, app=celery_app) — Inspects task state and retrieves stored result from backend

  result = AsyncResult(task_id, app=celery_app); status = result.status
  

• celery_app.control.revoke(task_id, terminate=True) — Cancels a queued task or terminates an actively running task

  celery_app.control.revoke("c3b9b4a1-8d2b-4e89", terminate=True)
  

• @celery_app.task(bind=True, max_retries=3, time_limit=60) — Decorator configuring bound context, retry limit, and execution timeout

  @celery_app.task(bind=True, max_retries=3, time_limit=60)
  def run_job(self): pass
  

• self.retry(exc=e, countdown=10) — Re-enqueues task after failure with an optional delay timer

  raise self.retry(exc=error, countdown=10)
  

Comparison Table

Common Pitfalls

• Mistake: Sharing the same Redis database number for both broker and result backend. Avoid: Set broker to database 0 and backend to database 1 in configuration URLs.
• Mistake: Starting the FastAPI server before starting the Redis broker and Celery workers. Avoid: Initialize components sequentially: start Redis first, launch Celery workers second, and start FastAPI last.
• Mistake: Calling a Celery task directly like a normal function inside an endpoint. Avoid: Call task.delay() or task.apply_async() to ensure execution runs on Celery workers.
• Mistake: Implementing client HTTP polling in a tight loop without a sleep delay. Avoid: Add a fixed wait interval like two seconds between status poll requests.

FAQs

• Why does Celery require both a message broker and a result backend? The broker acts as a temporary queue for distributing pending jobs to workers, while the backend stores task status and completed output for client retrieval.
• What states can Celery's AsyncResult return? Standard states include PENDING (queued/waiting), STARTED (processing), SUCCESS (completed), FAILURE (raised exception), and REVOKED (cancelled).
• Why isolate Redis databases for broker and result storage? Separating databases prevents key collisions, separates transient queue traffic from long-term result storage, and simplifies monitoring.
• When should I upgrade from HTTP polling to WebSockets? Upgrade to WebSockets when your user interface requires real-time task notifications without the latency or bandwidth overhead of recurring HTTP requests.