FastAPI Async Pitfalls: When to Use BackgroundTasks, run_in_executor, or Celery

Many developers choose FastAPI for its async capabilities, expecting out‑of‑the‑box ultra‑fast APIs. In practice:

Async is best suited for I/O‑intensive workloads such as network and database calls.

CPU‑intensive or blocking code (e.g., math, image/video processing, PDF generation, ML inference) blocks the event loop and harms every request.

Blocking code negates async benefits—connections may stall, leading to timeouts and poor scalability.

Why?

FastAPI runs on Starlette with an event‑driven loop (similar to Node.js). If a handler runs slow code that does not await or yield, it blocks every incoming request on that worker.

When to use BackgroundTasks and run_in_executor

These built‑in strategies address different problems:

BackgroundTasks

Executed after FastAPI sends the response.

Ideal for lightweight, non‑blocking post‑response work such as sending email.

Runs in the same thread as the request, so it is not suitable for heavy CPU work or truly asynchronous tasks.

Example:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(data: str):
    with open("log.txt", "a") as f:
        f.write(data)

@app.post("/log/")
async def log_endpoint(data: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, data)
    return {"message": "Logging scheduled"}

run_in_executor

Schedules blocking/CPU‑intensive work to a separate thread (or process).

Used for CPU‑heavy or blocking I/O code to prevent event‑loop starvation.

Example:

import asyncio

def do_cpu_heavy():
    # time‑consuming operation
    ...

@app.get("/heavy/")
async def cpu_bound():
    result = await asyncio.get_event_loop().run_in_executor(None, do_cpu_heavy)
    return {"result": result}

Passing None uses a thread pool; for true CPU‑bound tasks, use ProcessPoolExecutor because threads cannot bypass Python’s GIL.

Celery vs. Dramatiq for Task Offloading

Celery : Proven ecosystem, supports retries/failures, works with Redis or RabbitMQ; downside – complex configuration and requires worker processes.

Dramatiq : Simple setup, supports async/await, fits pure‑Python environments; downside – ecosystem less mature.

Celery excels at heavy tasks such as PDF generation, video processing, large‑scale ETL, or ML inference.

Task queues decouple workload from the web process, so slow jobs never block the API.

FastAPI endpoint → submit task to Celery (immediate 200/202 response).

Celery worker picks up and processes the task.

(Optional) Notify completion via webhook, WebSocket, or polling.

Handling Synchronous Libraries in Async Apps

Many popular libraries (e.g., Pandas, requests, PIL) are synchronous. To use them safely:

Always wrap with run_in_executor or offload to a task queue.

Never call slow synchronous code directly inside an async endpoint.

For large or truly CPU‑intensive workloads, prefer process‑based workers ( ProcessPoolExecutor or Celery) to avoid the GIL bottleneck.

Avoid Starlette Deadlocks

Never perform heavy computation, blocking calls, or sleep inside async routes.

For slow I/O, ensure the library is async or wrap it with run_in_executor.

Avoid spawning subprocesses from within the event‑loop callbacks, which can deadlock under uvicorn/gunicorn.

Real‑World Use Case: PDF Generation & Video Processing

Scenario

Provide a /generate-pdf/ endpoint.

Wrong Approach

from fpdf import FPDF

@app.post("/generate-pdf/")
async def generate_pdf():
    pdf = FPDF()  # heavy operation!
    pdf.add_page()
    pdf.output("report.pdf")
    return {"status": "done"}

This blocks the event loop, causing slow responses or timeouts.

Correct Approach

Submit PDF generation to a Celery/Dramatiq task.

Immediately return a task ID.

Let the client poll or receive a notification when done.

FastAPI + Celery Template

Production‑grade architecture:

main.py

from fastapi import FastAPI
from celery.result import AsyncResult
from tasks import create_pdf

app = FastAPI()

@app.post("/pdf/")
def create_pdf_endpoint(content: dict):
    task = create_pdf.delay(content)
    return {"task_id": task.id}

@app.get("/pdf_status/{task_id}")
def get_pdf_status(task_id: str):
    status = AsyncResult(task_id).status
    return {"status": status}

tasks.py

from celery import Celery

celery_app = Celery("tasks", broker="redis://localhost:6379/0")

@celery_app.task()
def create_pdf(content):
    # heavy CPU/PDF logic
    ...

Use Redis or RabbitMQ as the broker.

Start Celery with celery -A tasks worker --loglevel=info.

Conclusion

For lightweight, non‑blocking post‑request work, use FastAPI’s BackgroundTasks.

For CPU‑intensive or heavy blocking work, use run_in_executor or, better, a dedicated task queue such as Celery.

Always decouple slow business logic from API request handling – your users and team will thank you!