High-Throughput REST APIs with FastAPI, AsyncIO, and Connection Pooling

Python is often criticized for slower raw execution compared to compiled languages like Go or Rust. However, for 95% of web APIs, throughput bottlenecks are not caused by CPU math, but by waiting on I/O operations: waiting for database queries to return, waiting for third-party HTTP microservices, or waiting for disk reads.

In this engineering guide, we dissect how to build microservices capable of sustaining 10,000+ requests per second using FastAPI, the uvloop asynchronous event loop, Pydantic v2 serialization, and PostgreSQL connection pooling with asyncpg.

1. Demystifying the Python Event Loop: uvloop & AsyncIO

Standard Python web frameworks like Flask or Django (in WSGI mode) use a synchronous threading model: each incoming HTTP request occupies an entire operating system thread. Under heavy traffic (e.g., 1,000 concurrent visitors), your server exhausts memory and context-switching overhead brings throughput to a crawl.

FastAPI is built on top of Starlette and ASGI (Asynchronous Server Gateway Interface). When an asynchronous endpoint awaits a database query, Python releases the thread to handle hundreds of other concurrent incoming requests:

Synchronous Worker (Flask / WSGI):
[Request 1] ─── (Blocks thread for 50ms database query) ───> [Wait...] ───> [Done]
[Request 2] ─────────────────────── (Queued in backlog...) ───────────────>

Asynchronous Worker (FastAPI / ASGI):
[Request 1] ─── (Awaits query) ──────┐
[Request 2] ─── (Processes immediately) │──> Single thread handles 1,000+ I/O tasks
[Request 3] ─── (Returns response) ──┘

By substituting Python's standard asyncio loop with uvloop (a C-based implementation built on libuv, the engine behind Node.js), execution speed nearly doubles, rivaling Go and Node.js.

2. The Database Connection Pool Architecture

The most common mistake in FastAPI services is opening and closing database connections on every incoming request. Establishing a new TLS/TCP connection to PostgreSQL incurs a 30–80ms penalty per query.

We use asyncpg—the fastest asynchronous PostgreSQL driver in existence—with an optimized connection pool managed through FastAPI's lifespan events.

Complete Battle-Tested Architecture

# app/main.py
from contextlib import asynccontextmanager
import asyncpg
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel, Field
import os

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/production_db")

# Global connection pool reference
db_pool: asyncpg.Pool = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Initialize persistent database pool
    global db_pool
    print("🚀 Initializing asyncpg connection pool...")
    db_pool = await asyncpg.create_pool(
        dsn=DATABASE_URL,
        min_size=10,             # Keep 10 idle connections warm
        max_size=50,             # Peak concurrent connections
        max_queries=50000,       # Recycle connection after 50k queries to prevent leaks
        max_inactive_connection_lifetime=300.0,
        timeout=10.0
    )
    yield
    # Shutdown: Gracefully terminate connections
    print("🛑 Draining asyncpg connection pool...")
    await db_pool.close()

app = FastAPI(
    title="High-Throughput Analytics Engine",
    lifespan=lifespan,
    docs_url="/docs"
)

# Dependency injection for zero-overhead connection acquisition
async def get_db_connection():
    async with db_pool.acquire() as connection:
        yield connection

# Pydantic v2 Model (Compiled in Rust for instant serialization)
class EventPayload(BaseModel):
    user_id: str = Field(..., min_length=3, max_length=64)
    event_type: str = Field(..., min_length=2, max_length=32)
    payload_data: dict = Field(default_factory=dict)

class EventResponse(BaseModel):
    id: int
    status: str

@app.post("/api/v1/events", response_model=EventResponse, status_code=status.HTTP_201_CREATED)
async def record_event(
    event: EventPayload, 
    conn: asyncpg.Connection = Depends(get_db_connection)
):
    query = """
        INSERT INTO user_activity_events (user_id, event_type, metadata, created_at)
        VALUES ($1, $2, $3, NOW())
        RETURNING id;
    """
    try:
        record_id = await conn.fetchval(
            query, 
            event.user_id, 
            event.event_type, 
            event.payload_data
        )
        return {"id": record_id, "status": "recorded"}
    except Exception as exc:
        raise HTTPException(status_code=500, detail="Database insertion failure")

@app.get("/healthz")
async def health_check():
    return {"status": "healthy", "pool_free": db_pool.get_idle_size()}

3. Avoiding the CPU-Bound Trap in AsyncIO

In an asynchronous application, running a synchronous, CPU-intensive operation (such as image resizing, password hashing via bcrypt, or heavy CSV parsing) blocks the entire event loop, freezing all other concurrent connections.

Offloading CPU Work to Thread Pools

import asyncio
from concurrent.futures import ThreadPoolExecutor
import hashlib

# Dedicated thread pool for compute-heavy tasks
cpu_executor = ThreadPoolExecutor(max_workers=8)

def compute_heavy_hash(data: bytes) -> str:
    # Simulating intensive cryptography / CPU work
    result = data
    for _ in range(100_000):
        result = hashlib.sha256(result).digest()
    return result.hex()

@app.post("/api/v1/process-hash")
async def process_hash_endpoint(payload: bytes):
    loop = asyncio.get_running_loop()
    # Execute non-blocking on separate worker thread
    digest = await loop.run_in_executor(cpu_executor, compute_heavy_hash, payload)
    return {"hash": digest}

4. Production Deployment with Uvicorn & Gunicorn

Never run uvicorn main:app --reload in production. Deploy using Gunicorn as a process manager supervising multiple Uvicorn worker processes:

# Production command: 2-4 workers per CPU core
gunicorn app.main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --backlog 2048 \
  --timeout 30 \
  --keep-alive 5 \
  --access-logfile -

5. Concurrency Load Test Benchmarks (wrk)

Benchmark testing executed using wrk with 12 threads and 500 concurrent connections over 30 seconds:

FastAPI with uvloop delivers 10x higher throughput than traditional synchronous Python frameworks, closely matching compiled Go microservices.

Frequently Asked Questions (FAQ)

Should I define endpoints as async def or standard def?

If your endpoint performs asynchronous I/O (await db.fetch(), await client.get()), define it as async def. If your endpoint uses synchronous, blocking libraries (like standard requests or psycopg2), define it as standard def. FastAPI will automatically offload standard def functions to an internal thread pool to prevent blocking the event loop.

Why is asyncpg faster than SQLAlchemy?

asyncpg implements the PostgreSQL binary protocol directly in Cython/C with zero intermediary translation layers. While SQLAlchemy provides a comprehensive ORM with migration tooling, asyncpg raw queries can execute 3x to 5x faster for high-volume endpoints.

What is the ideal database pool size?

A good formula is: (CPU_cores * 2) + disk_spindles. For a 4-core database server, a pool of 15–30 connections usually delivers maximum throughput. Setting pool sizes to hundreds of connections actually degrades PostgreSQL performance due to CPU lock contention and memory paging.