Asynchronous Workflows with FastAPI, Celery, and Redis: Handling Heavy Batch Computations

The Golden Rule of Web APIs

Never execute CPU-bound computation or unbounded network I/O inside an HTTP request handler. When an HTTP client connects to your server, an open socket occupies connection pools and worker threads. Holding that connection open for 60 seconds while processing data exhausts web server capacity and causes gateway timeouts.

The Asynchronous Task Architecture

[ Web Client ] 
      │  1. POST /v1/jobs (Submit heavy task)
      ▼
┌──────────────────┐  2. Push Task    ┌──────────────────┐
│ FastAPI Gateway  │─────────────────▶│   Redis Broker   │
└────────┬─────────┘                  └────────┬─────────┘
         │ 3. Return 202 Accepted              │ 4. Pull Task
         │    {"job_id": "uuid-123"}           ▼
         ▼                            ┌──────────────────┐
[ Web Client ]                        │  Celery Worker   │
      │                               │  (Heavy Compute) │
      │  5. GET /v1/jobs/uuid-123     └────────┬─────────┘
      │     (Poll Status / Websocket)          │ 5. Save Result
      ▼                                        ▼
┌──────────────────┐                  ┌──────────────────┐
│ FastAPI Gateway  │◀─────────────────│ Redis Result DB  │
└──────────────────┘                  └──────────────────┘

FastAPI Job Submission Endpoint

from fastapi import FastAPI, status
from pydantic import BaseModel
from celery import Celery

app = FastAPI()
celery_app = Celery("tasks", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")

@app.post("/v1/export-report", status_code=status.HTTP_202_ACCEPTED)
async def trigger_export(payload: dict):
    # Enqueue task onto Redis broker without blocking
    task = celery_app.send_task("tasks.compute_heavy_report", args=[payload])
    return {"job_id": task.id, "status": "processing", "poll_url": f"/v1/jobs/{task.id}"}

@app.get("/v1/jobs/{job_id}")
async def check_job_status(job_id: str):
    res = celery_app.AsyncResult(job_id)
    if res.state == "PENDING":
        return {"status": "pending", "ready": False}
    elif res.state == "SUCCESS":
        return {"status": "completed", "ready": True, "result": res.result}
    elif res.state == "FAILURE":
        return {"status": "failed", "error": str(res.info)}
    return {"status": res.state, "ready": False}