Async Python & Background Tasks

FastAPI is built on asyncio — Python's async I/O framework. Understanding async is critical for writing performant FastAPI code.

The core insight: when your code does I/O (DB query, HTTP call, file read), the CPU is idle waiting. In synchronous code, this blocks the entire thread. In async code, you yield control back to the event loop so it can process other requests while you wait.

async/await:
• async def → defines a coroutine (a function that can be paused)
• await → pauses the coroutine until the awaited thing completes, without blocking the thread
• You must await only async functions from async-compatible libraries

When to use async vs sync in FastAPI:
• Use async def when calling async libraries (httpx, asyncpg, aiofiles, aioredis)
• Use def (sync) when calling sync libraries (requests, psycopg2, PIL) — FastAPI runs sync routes in a threadpool automatically so they don't block the event loop
• Never call a sync blocking function from async def — it will freeze your entire server

Background Tasks run after the response is sent to the client:
• Client gets 200 immediately
• Task runs after (send email, write log, trigger processing)
• Uses FastAPI's BackgroundTasks, not asyncio.create_task
• For heavy jobs (video encoding, bulk email), use Celery + Redis instead

asyncio.gather runs multiple coroutines concurrently — perfect for fetching from multiple APIs in parallel.

1from fastapi import FastAPI, BackgroundTasks
2import asyncio
3import httpx
4import time
5
6app = FastAPI()
7
8# ── Async route — non-blocking ─────────────────────────
9@app.get("/async-data")
10async def get_data():
11    async with httpx.AsyncClient() as client:
12        # does NOT block the event loop while waiting
13        response = await client.get("https://httpbin.org/delay/1")
14    return {"data": response.json()}
15
16# ── Sync route — FastAPI runs in threadpool (ok) ───────
17@app.get("/sync-data")
18def get_data_sync():
19    # sync libraries are fine in sync def routes
20    time.sleep(0.1)
21    return {"data": "sync but ok"}
22
23# ── Concurrent requests with asyncio.gather ────────────
24@app.get("/parallel")
25async def parallel_fetch():
26    async with httpx.AsyncClient() as client:
27        # both run at the same time!
28        results = await asyncio.gather(
29            client.get("https://httpbin.org/get"),
30            client.get("https://httpbin.org/uuid"),
31        )
32    return {"results": [r.json() for r in results]}
33
34# ── Background tasks ───────────────────────────────────
35def write_log(message: str):
36    """This runs AFTER the response is sent"""
37    with open("activity.log", "a") as f:
38        f.write(f"{message}
39")
40
41def send_welcome_email(email: str, username: str):
42    """Simulate sending email (heavy task, runs in background)"""
43    time.sleep(2)  # simulate slow email sending
44    print(f"Email sent to {email} for {username}")
45
46@app.post("/users/", status_code=201)
47def create_user(username: str, email: str, background_tasks: BackgroundTasks):
48    # ... create user in db ...
49    background_tasks.add_task(write_log, f"New user: {username}")
50    background_tasks.add_task(send_welcome_email, email, username)
51
52    # Response goes to client immediately — email sends after
53    return {"message": "User created", "username": username}