Async

from fastapi import FastAPI

app = FastAPI()

# Sync — FastAPI runs this in a threadpool automatically
@app.get("/sync")
def sync_route():
    result = blocking_db_call()   # OK here — not blocking the event loop
    return result

# Async — runs on the event loop; NEVER block here
@app.get("/async")
async def async_route():
    result = await async_db_call()
    return result

import httpx
import asyncio

@app.get("/external")
async def call_external():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/data")
        return resp.json()

@app.get("/parallel")
async def parallel():
    # Run two awaitable calls concurrently:
    result_a, result_b = await asyncio.gather(
        fetch_a(),
        fetch_b(),
    )
    return {"a": result_a, "b": result_b}

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

async def get_token_data(token: str = Depends(oauth2_scheme)):
    payload = await async_verify_token(token)
    return payload

@app.get("/me")
async def me(data = Depends(get_token_data)):
    return data

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

@app.get("/cpu-heavy")
async def cpu_heavy():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, blocking_cpu_function, arg)
    return {"result": result}

# Shortcut with anyio:
import anyio

@app.get("/blocking")
async def blocking():
    result = await anyio.to_thread.run_sync(sync_blocking_func)
    return result

@app.get("/redis")
async def redis_route():
    async with aioredis.from_url("redis://localhost") as redis:
        value = await redis.get("key")
        return {"value": value}

from contextlib import asynccontextmanager
import httpx

http_client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client
    http_client = httpx.AsyncClient()
    yield
    await http_client.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/fetch")
async def fetch(url: str):
    resp = await http_client.get(url)
    return resp.json()

# asyncio.gather — cancel all on first error
async def route():
    a, b, c = await asyncio.gather(task_a(), task_b(), task_c())
    return [a, b, c]

# asyncio.TaskGroup (Python 3.11+) — structured concurrency
async def route():
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(fetch_a())
        task_b = tg.create_task(fetch_b())
    return [task_a.result(), task_b.result()]

# pip install aiofiles
import aiofiles

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("data.txt", "r") as f:
        content = await f.read()
    return {"content": content}

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    while True:
        data = await ws.receive_text()
        await ws.send_text(f"Echo: {data}")

from fastapi.responses import StreamingResponse

async def event_generator():
    for i in range(100):
        await asyncio.sleep(0.1)
        yield f"data: {i}\n\n"

@app.get("/sse")
async def sse():
    return StreamingResponse(event_generator(), media_type="text/event-stream")