Skip to main content

Module 6 — Async requests and background tasks

FastAPI is built on ASGI, so the words async def appear all over its documentation. It is tempting to assume that writing every route async makes an ML API faster. This module explains why the opposite is often true, gives one concrete pattern for the ML case (a blocking model call handed off to a thread pool), and then covers the two shapes of "do it in the background": FastAPI's built-in BackgroundTasks for cheap work, and a real worker queue for anything that needs to survive a restart. The running example gains a route that accepts a CSV of subscribers and returns a job ID.

async in one paragraph, for the ML case​

An async def route runs on the event loop and yields cooperatively whenever it awaits I/O. This is enormously useful when the route spends most of its time waiting for a database, an HTTP call, or a socket — the same worker can handle another request during the wait. It is actively harmful when the route spends its time inside a CPU-bound Python call, because the event loop is single-threaded and the call blocks every other request on that worker until it returns. A model.predict() on scikit-learn or on a bare PyTorch model is CPU-bound. Putting it directly in an async def route is the most common performance regression of newly rewritten ML services.

The correct pattern is one of the two below:

  • Sync route + Uvicorn workers. Write the route as a plain def. Starlette runs sync routes in a thread pool (AnyIO), so the event loop stays free. This is the pattern used in every module of this course so far, and it is the pattern to keep for anything that calls a CPU-bound model.
  • async route + explicit thread pool for the model call. Use async def if the route needs to await something (a feature store HTTP call, for example), and offload the model call to a thread with run_in_threadpool:
# app/main.py
from fastapi import FastAPI
from starlette.concurrency import run_in_threadpool

from app.schemas import Subscriber, Prediction
from app.scoring import _score_one
from app.features import fetch_features  # async HTTP call to a feature store


app = FastAPI(title="Churn scoring API", version="0.6.0")


@app.post("/predict/enriched", response_model=Prediction, tags=["prediction"])
async def predict_enriched(subscriber: Subscriber) -> Prediction:
    # Await I/O: the event loop can serve other requests during this call.
    enriched = await fetch_features(subscriber)
    # Offload CPU work to a thread; the loop stays free for other requests.
    return await run_in_threadpool(_score_one, enriched)

Two lines carry the module's central lesson. await fetch_features(...) is exactly what async was designed for. await run_in_threadpool(_score_one, ...) protects the loop from a call that would otherwise block it for tens or hundreds of milliseconds. Mixing the two shapes is what an async ML route should look like — never a raw synchronous model.predict inside async def.

When async genuinely helps​

The single most useful case for async in an ML API is fan-out I/O: fetching many features from many stores in parallel, or calling several models and aggregating their outputs. Structured concurrency with asyncio.TaskGroup (Python 3.11+) is the idiomatic tool:

import asyncio


async def gather_features(subscriber_id: str) -> dict:
    async with asyncio.TaskGroup() as tg:
        billing = tg.create_task(billing_features(subscriber_id))
        usage = tg.create_task(usage_features(subscriber_id))
        support = tg.create_task(support_features(subscriber_id))
    return {**billing.result(), **usage.result(), **support.result()}

Three HTTP calls that would each take 40 ms and would run sequentially in ~120 ms total now run concurrently in ~40 ms. That is a real p50 gain, and it is worth the async shape.

BackgroundTasks: fire-and-forget after the response​

FastAPI's BackgroundTasks runs a Python callable after the response is sent, on the same process. It is the right tool for small, best-effort side effects that the caller does not need to wait for: writing an audit line, sending a webhook, invalidating a cache:

from fastapi import BackgroundTasks


def audit(subscriber_id: str, probability: float) -> None:
    with open("/var/log/churn/audit.log", "a", encoding="utf-8") as f:
        f.write(f"{subscriber_id}\t{probability}\n")


@app.post("/predict", response_model=Prediction, tags=["prediction"])
def predict(subscriber: Subscriber, background: BackgroundTasks) -> Prediction:
    prediction = _score_one(subscriber)
    background.add_task(audit, subscriber.subscriber_id, prediction.churn_probability)
    return prediction

Two properties matter and are often misunderstood. BackgroundTasks runs in the same process as the request, so a crash of the process kills the task. And there is no retry, no queue, no visibility: if the audit write fails, the API caller has already received a 200 and there is nothing to try again. Use it for work whose loss you can accept.

File scoring: the shape that pays for itself​

The batch route from module 4 caps at 500 rows. A Streamlit CSV upload with 20 000 subscribers is legitimate work that does not fit in a synchronous request — the client would hit its timeout, workers would be pinned for seconds, and any retry would double the load. The right shape is a two-step protocol: accept the file, return a job ID; the client polls a status route.

# app/jobs.py
import uuid
from pathlib import Path

JOBS: dict[str, dict] = {}  # in-memory; a real deployment uses Redis or a DB.
JOBS_DIR = Path("/tmp/churn-jobs"); JOBS_DIR.mkdir(exist_ok=True)


def submit(file_bytes: bytes) -> str:
    job_id = str(uuid.uuid4())
    JOBS[job_id] = {"status": "queued", "rows_processed": 0, "rows_total": None}
    (JOBS_DIR / f"{job_id}.csv").write_bytes(file_bytes)
    return job_id


def status(job_id: str) -> dict | None:
    return JOBS.get(job_id)

The route itself accepts an upload and schedules the work:

# app/main.py (continued)
from fastapi import FastAPI, File, UploadFile, BackgroundTasks, HTTPException, status

from app.jobs import submit, status as job_status
from app.scoring_batch import score_csv_job


@app.post("/predict/file", tags=["prediction"], status_code=status.HTTP_202_ACCEPTED)
def predict_file(background: BackgroundTasks, file: UploadFile = File(...)) -> dict:
    if not file.filename.endswith(".csv"):
        raise HTTPException(400, {"code": "unsupported_type", "message": "CSV only"})
    job_id = submit(file.file.read())
    background.add_task(score_csv_job, job_id)  # runs after the response
    return {"job_id": job_id, "status": "queued"}


@app.get("/predict/file/{job_id}", tags=["prediction"])
def predict_file_status(job_id: str) -> dict:
    st = job_status(job_id)
    if st is None:
        raise HTTPException(404, {"code": "job_not_found", "message": job_id})
    return {"job_id": job_id, **st}

The 202 Accepted status code is the accurate signal to the caller: "your request is valid and I will process it later". The polling loop on the client hits GET /predict/file/{job_id} every second or two and reads a growing rows_processed counter until status becomes done.

When BackgroundTasks stops being enough​

score_csv_job above runs in the same process as the API. Three things break as soon as the traffic is real:

  • Restarts lose work. A deploy or a crash halfway through a 20 000-row job drops the job silently; JOBS[job_id] was in memory.
  • Workers get starved. A ten-minute CSV job pins one of your Uvicorn workers for ten minutes, and if you only have four workers, one busy job costs you a quarter of your capacity.
  • No horizontal scaling. Every replica has its own JOBS dict; a client that polls a different replica than the one that accepted the file gets a 404.

The answer is a real worker queue. In the Python world, the two standard options are Celery with Redis or RabbitMQ, and RQ (simpler, Redis-only). The service becomes a producer that writes a job to a queue, and a separate pool of worker processes consumes it:

# app/queue.py — preview only
from celery import Celery

celery = Celery("churn", broker="redis://redis:6379/0", backend="redis://redis:6379/1")


@celery.task(name="score_csv")
def score_csv_task(job_id: str) -> None:
    # Same body as score_csv_job, but runs in a worker process.
    ...

The API pod is now free within milliseconds of accepting the file; the worker pool scales independently based on queue depth. Job state moves from JOBS to Redis, which survives restarts of both API and workers. This is where a real ML API converges by month two of production, and the two-step protocol above is exactly the interface that migration keeps intact — the callers do not know or care whether the work runs in the same process or on a dedicated worker.

async def around a blocking model call is a latency bomb

The single most common regression when a team "adopts async" is to write async def predict(...) and then call model.predict(...) synchronously inside. The event loop is now blocked for the duration of the call, and every other request on that worker waits. Either keep the route sync, or wrap the model call in run_in_threadpool. A tiny benchmark under load makes the difference obvious.

In summary​

  • async def helps when the route awaits I/O; it hurts when it wraps a CPU-bound model call. Keep sync routes for pure model paths, or wrap the model call in run_in_threadpool.
  • Use asyncio.TaskGroup to fan-out I/O to feature stores and other models — that is where async produces a real p50 gain.
  • Use BackgroundTasks for best-effort side effects (audit log, webhook) that share the request's process; accept that a crash loses them.
  • For file-scoring or any minute-long work, return 202 Accepted with a job ID and process asynchronously. Move to a real worker queue (Celery, RQ) as soon as scale, restarts or horizontal replicas start hurting.

Next module: token authentication — protecting the routes of this module with an API key and a security dependency, and previewing JWT and rate limiting.