FastAPI load performance

import asyncio
import time
from contextlib import asynccontextmanager

import aiosqlite
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel

DB_PATH = "./bench.db"
BIG_SQL = "SELECT id, user_id, payload FROM events ORDER BY id DESC LIMIT ?"


class Event(BaseModel):
    id: int
    user_id: int
    payload: str


class Pool:
    """Fixed-size pool, built once at startup, never mid-request."""

    def __init__(self, path: str, size: int):
        self.path, self.size = path, size
        self._q: asyncio.Queue = asyncio.Queue(size)

    async def open(self):
        for _ in range(self.size):
            conn = await aiosqlite.connect(self.path)
            await conn.execute("PRAGMA journal_mode=WAL")
            await conn.execute("PRAGMA synchronous=NORMAL")
            await conn.execute(
                "CREATE TABLE IF NOT EXISTS events "
                "(id INTEGER PRIMARY KEY, user_id INTEGER, payload TEXT)"
            )
            await conn.commit()
            self._q.put_nowait(conn)

    async def close(self):
        while not self._q.empty():
            await self._q.get().close()

    def acquire(self):
        return self._q.get()

    def release(self, conn):
        self._q.put_nowait(conn)


pool = Pool(DB_PATH, size=8)


@asynccontextmanager
async def lifespan(_: FastAPI):
    await pool.open()
    yield
    await pool.close()


app = FastAPI(lifespan=lifespan)


async def fetch(limit: int):
    conn = await pool.acquire()
    try:
        cur = await conn.execute(BIG_SQL, (limit,))
        rows = await cur.fetchall()
        cols = [d[0] for d in cur.description]
        return [dict(zip(cols, r)) for r in rows]
    finally:
        pool.release(conn)


# 1. small result, no response_model
@app.get("/read/hot")
async def read_hot():
    return await fetch(20)


# 2. large result through Pydantic
@app.get("/read/validated", response_model=list[Event])
async def read_validated():
    return await fetch(500)


# 3. large result, FastAPI does no conversion at all
@app.get("/read/raw")
async def read_raw():
    return JSONResponse(await fetch(500))


# 4. blocking work in a SYNC handler: the anyio threadpool, 40 slots by default
@app.get("/slow/sync")
def slow_sync():
    time.sleep(0.05)
    return {"ok": True}


# 5. the same latency budget, expressed asynchronously
@app.get("/slow/async")
async def slow_async():
    await asyncio.sleep(0.05)
    return {"ok": True}


# 6. CPU-bound work inside an async handler: blocks the one loop
@app.get("/mixed")
async def mixed():
    total = 0
    for i in range(2_000_000):
        total += i
    return {"ok": total > 0}

import asyncio
import statistics
import sys
import time

import httpx

BASE = "http://127.0.0.1:8000"


async def worker(client, path, deadline, lat, errors):
    while time.monotonic() < deadline:
        start = time.perf_counter()
        try:
            r = await client.get(path)
            r.raise_for_status()
            lat.append((time.perf_counter() - start) * 1000)
        except Exception as e:  # noqa: BLE001
            errors.append(repr(e))


async def run(path, concurrency, duration):
    limits = httpx.Limits(max_connections=concurrency,
                          max_keepalive_connections=concurrency)
    async with httpx.AsyncClient(base_url=BASE, limits=limits, timeout=10.0) as client:
        lat, errors = [], []
        deadline = time.monotonic() + duration
        # warmup, untimed
        await asyncio.gather(*(client.get(path) for _ in range(30)))
        await asyncio.gather(
            *(worker(client, path, deadline, lat, errors) for _ in range(concurrency))
        )

    lat.sort()
    if not lat:
        print("all requests failed:", errors[:3])
        return
    q = int(0.99 * (len(lat) - 1))
    print(
        f"{path:16} conc={concurrency:3} rps={len(lat) / duration:8.1f} "
        f"p50={statistics.median(lat):6.2f}ms p99={lat[q]:6.2f}ms "
        f"fail={len(errors)}/{len(lat) + len(errors)}"
    )


if __name__ == "__main__":
    asyncio.run(run(
        sys.argv[1] if len(sys.argv) > 1 else "/read/hot",
        int(sys.argv[2]) if len(sys.argv) > 2 else 50,
        float(sys.argv[3]) if len(sys.argv) > 3 else 10.0,
    ))

pip install "fastapi[standard]" aiosqlite httpx
python -m uvicorn app:app --host 127.0.0.1 --port 8000 --workers 1 --loop uvloop

# concurrency sweep, one client process (watch it saturate)
for c in 1 8 24 64; do python loadgen.py /read/hot $c 10; done

# the sync-vs-async comparison: 8 clients x 8 concurrent = 64 offered
for i in $(seq 8); do python loadgen.py /slow/async 8 5 & done; wait
for i in $(seq 8); do python loadgen.py /slow/sync 8 5 & done; wait

supported concurrent users = (system RPS x user think time in seconds) / requests per page load