Building async APIs in FastAPI is straightforward, but production workloads reveal the importance of connection management. At scale, concurrency patterns that worked in development need refinement to handle database connection efficiency, request timing, and system stability under load. The difference between a FastAPI app that handles 100 requests per second and one that handles 10,000 often comes down to three decisions: how you manage connections, how you handle concurrent database access, and how you structure backpressure into your request pipeline.
This guide walks through the patterns that matter, with practical code examples and the reasoning behind each choice.
Why Async Matters for API Scalability
FastAPI runs on ASGI servers like Uvicorn, which use a single event loop per worker process to handle multiple concurrent requests. When a request awaits an I/O operation (database query, HTTP call, file read), the event loop pauses that request and processes another one. This is fundamentally different from sync frameworks like Flask, where each request gets its own thread or process.
The benefit is clear: with async/await, you handle thousands of concurrent connections using just a handful of worker processes. A single Uvicorn worker with 10 concurrent requests doesn’t need 10 threads. It needs one event loop and the ability to yield control when I/O blocks.
The key: this efficiency works best when your I/O operations are truly async. If you block the event loop with CPU work or synchronous I/O, you starve other requests. Database connection pooling becomes critical because every database connection that’s held open but idle is a resource you can’t use for another request.
Connection Pooling: The Foundation
Database performance in FastAPI applications depends heavily on connection management. Common approaches like creating a new connection per request or unlimited connections can lead to resource constraints. Connection pooling solves this by maintaining a fixed set of reusable connections, handing them out to requests, and returning them when done. For async code, use asyncpg (for PostgreSQL) or a pool-aware driver.
Here’s a minimal setup with asyncpg:
import asyncpg
from contextlib import asynccontextmanager
from fastapi import FastAPI
class Database:
def __init__(self, dsn: str, min_size: int = 10, max_size: int = 20):
self.dsn = dsn
self.min_size = min_size
self.max_size = max_size
self.pool = None
async def connect(self):
self.pool = await asyncpg.create_pool(
self.dsn,
min_size=self.min_size,
max_size=self.max_size,
command_timeout=60,
)
async def disconnect(self):
if self.pool:
await self.pool.close()
@asynccontextmanager
async def acquire(self):
async with self.pool.acquire() as connection:
yield connection
db = Database("postgresql://user:password@localhost/dbname")
@asynccontextmanager
async def lifespan(app: FastAPI):
await db.connect()
yield
await db.disconnect()
app = FastAPI(lifespan=lifespan)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
async with db.acquire() as conn:
user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
return user
Here’s what each parameter controls:
- min_size: Connections to maintain open at all times, even if unused. Set this to the baseline load you expect.
- max_size: Maximum connections allowed. Beyond this, requests queue. Set based on your database’s max_connections limit, typically leaving headroom for other services.
- command_timeout: Seconds to wait for a query before canceling. Prevents hung connections from blocking the pool.
Typical production settings for a moderate API: min_size=10, max_size=20 per worker. If you run 4 workers, that’s 80 connections total. Verify your database allows this.
Handling Concurrent Requests Correctly
Async code can be deceptive. This looks correct and works fine:
@app.get("/posts")
async def get_posts():
async with db.acquire() as conn:
posts = await conn.fetch("SELECT * FROM posts LIMIT 100")
return posts
The issue appears when you fetch related data inside a loop:
@app.get("/posts-with-authors")
async def get_posts_with_authors():
async with db.acquire() as conn:
posts = await conn.fetch("SELECT * FROM posts LIMIT 100")
for post in posts:
author = await conn.fetchrow("SELECT * FROM users WHERE id = $1", post['user_id'])
post['author'] = author
return posts
This approach is sequential: fetch 100 posts, then query the database 100 times in series for authors. This pattern holds the connection during I/O waits. A better approach is concurrent queries. Use asyncio.gather to run queries concurrently:
import asyncio
@app.get("/posts-with-authors")
async def get_posts_with_authors():
async with db.acquire() as conn:
posts = await conn.fetch("SELECT * FROM posts LIMIT 100")
async def fetch_author(post):
author = await conn.fetchrow("SELECT * FROM users WHERE id = $1", post['user_id'])
post['author'] = author
await asyncio.gather(*[fetch_author(post) for post in posts])
return posts
Even better, use a SQL join and avoid the N+1 query entirely:
@app.get("/posts-with-authors")
async def get_posts_with_authors():
async with db.acquire() as conn:
posts = await conn.fetch("""
SELECT p.*, u.name as author_name, u.email as author_email
FROM posts p
LEFT JOIN users u ON p.user_id = u.id
LIMIT 100
""")
return posts
This is the real optimization: smart queries beat concurrent queries. But when you do need concurrency, asyncio.gather is your tool.
Backpressure and Request Queuing
Backpressure is the practice of refusing or delaying requests when the system is overloaded, rather than accepting them and degrading. Without it, a traffic spike causes all workers to queue database connections, requests pile up, memory grows, and everything gets slow.
The simplest backpressure is built into connection pooling: when the pool is exhausted, new requests wait. But you can be more explicit:
from fastapi import HTTPException, status
import time
class PoolMonitor:
def __init__(self, pool):
self.pool = pool
self.rejection_threshold = 0.9 # Reject if pool is 90% full
async def check_capacity(self):
size = self.pool.get_size()
free = self.pool.get_idle_size()
utilization = (size - free) / size if size > 0 else 0
if utilization > self.rejection_threshold:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Service overloaded, try again later"
)
monitor = PoolMonitor(db.pool)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
await monitor.check_capacity()
async with db.acquire() as conn:
user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
return user
This returns 503 when the pool is saturated, signaling to clients (and load balancers) to back off. Clients with exponential backoff will retry. Those without will fail fast, reducing load on the system.
Concurrency Limits and Semaphores
Sometimes you need to limit concurrency at the application level, not just the connection pool. For example, if an external API has rate limits, you can use asyncio.Semaphore:
import asyncio
class RateLimitedClient:
def __init__(self, max_concurrent: int = 5):
self.semaphore = asyncio.Semaphore(max_concurrent)
async def call_external_api(self, endpoint: str):
async with self.semaphore:
# This block runs with max 5 concurrent calls
# Other calls wait
result = await fetch(f"https://api.example.com{endpoint}")
return result
client = RateLimitedClient(max_concurrent=5)
@app.get("/enriched-data/{id}")
async def get_enriched_data(id: int):
async with db.acquire() as conn:
local_data = await conn.fetchrow("SELECT * FROM data WHERE id = $1", id)
external_data = await client.call_external_api(f"/data/{id}")
return {"local": local_data, "external": external_data}
Semaphores are lightweight and perfect for this use case. They don’t consume connections or threads; they simply coordinate when code blocks can run.
Timeouts and Cancellation
Long-running queries or external API calls can hang indefinitely. Always set timeouts:
import asyncio
from fastapi import HTTPException, status
@app.get("/slow-endpoint")
async def slow_endpoint():
try:
async with db.acquire() as conn:
result = await asyncio.wait_for(
conn.fetch("SELECT * FROM large_table WHERE condition = true"),
timeout=5.0
)
except asyncio.TimeoutError:
raise HTTPException(
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
detail="Query took too long"
)
return result
The connection is returned to the pool even if the query times out. The timeout also cancels the underlying database operation (in most drivers), freeing server-side resources.
Comparing Sync vs Async Performance
A practical comparison: a sync Flask app with 4 workers (each with 4 threads) can handle roughly 16 concurrent requests. A single Uvicorn worker with async/await can handle thousands. But the real difference emerges under realistic workloads.
Scenario: 100 concurrent users, each making requests that query a database (10ms latency) and call an external API (100ms latency).
Sync app with 4 workers and 4 threads each: 16 concurrent requests. The 100 users queue. Throughput is limited by thread count, not by the I/O latency.
Async app with 4 workers: each worker handles many concurrent requests. When one request awaits the database, the worker processes another. Throughput is much higher because workers aren’t blocked waiting for I/O.
Async code requires attention to detail. Blocking calls (synchronous database drivers, CPU-intensive computation, synchronous file I/O) can impact the event loop’s ability to process other requests. Sync code is more predictable but doesn’t scale as far.
Production Optimization Checklist
Before deploying a FastAPI application to production, verify:
- Connection pool sizing: min_size and max_size match your workload and database limits.
- Query timeouts: All database queries have explicit timeouts to prevent hung connections.
- No blocking operations: Audit for synchronous I/O, CPU-heavy loops, or blocking libraries in async functions.
- Backpressure: The application rejects or delays requests when overloaded, rather than queuing indefinitely.
- Monitoring: Track pool utilization, query times, and request queue depth. Set alerts for saturation.
- Graceful shutdown: Close the connection pool and cancel in-flight requests when shutting down.
- Database prepared statements: Use parameterized queries (asyncpg does this by default) to prevent SQL injection and improve caching.
Monitoring and Debugging
Add visibility into your connection pool and request pipeline:
from prometheus_client import Counter, Histogram, Gauge
import time
# Metrics
pool_size = Gauge('db_pool_size', 'Current pool size')
pool_idle = Gauge('db_pool_idle', 'Idle connections')
query_duration = Histogram('db_query_seconds', 'Query duration')
query_errors = Counter('db_query_errors_total', 'Query errors')
@app.middleware("http")
async def log_pool_stats(request, call_next):
if db.pool:
pool_size.set(db.pool.get_size())
pool_idle.set(db.pool.get_idle_size())
start = time.time()
try:
response = await call_next(request)
except Exception as e:
query_errors.inc()
raise
finally:
duration = time.time() - start
query_duration.observe(duration)
return response
Export these metrics to Prometheus and visualize in Grafana. Watch for pool saturation, query latency spikes, and error rates. These metrics tell you when to scale horizontally (add workers) or optimize queries.
Conclusion
High-performance async APIs aren’t built on raw concurrency. They’re built on thoughtful connection management, explicit timeouts, and backpressure. FastAPI and asyncpg give you the tools; the art is using them correctly.
Start with a properly sized connection pool, add backpressure when the pool saturates, and monitor relentlessly. Optimize queries before scaling horizontally. Test under realistic load before production, and be prepared to adjust pool sizes based on real traffic patterns.
The patterns in this guide apply to any async framework. The specific syntax changes, but the principles remain: manage connections as a limited resource, respect timeouts, and give your system a way to say no when it’s full.
What is the difference between min_size and max_size in asyncpg connection pooling?
min_size is the number of connections the pool maintains at all times, even if unused. max_size is the maximum number of connections allowed. When the pool reaches max_size, new requests wait for a connection to be returned. Set min_size to your baseline expected concurrency and max_size higher to handle traffic spikes. Going too high wastes database resources and can exhaust the server’s connection limit.
Why does my FastAPI app slow down under load even with async/await?
The most common cause is connection pool exhaustion. If all connections are in use and new requests queue, latency increases. Other causes include N+1 queries (querying in loops), blocking operations (synchronous I/O or CPU work), or queries without timeouts that hang indefinitely. Monitor pool utilization and query times to identify the bottleneck.
Should I use connection pooling with SQLAlchemy async or asyncpg directly?
Both work, but asyncpg is lower-level and gives you more control. SQLAlchemy async adds an ORM layer and manages pooling for you, which is convenient but has overhead. For maximum performance, use asyncpg directly. For rapid development, SQLAlchemy async is reasonable if you’re aware of the trade-offs.
How do I prevent a single slow query from blocking other requests?
Use asyncio.wait_for with a timeout to cancel queries that exceed a threshold. Also, ensure each query gets its own connection from the pool and returns it promptly. Avoid holding connections open longer than necessary, and use asyncio.gather to run independent queries concurrently instead of sequentially.
Is async Python always faster than sync Python for APIs?
Async is faster for I/O-bound workloads (database queries, HTTP calls) because it avoids the overhead of threads. For CPU-bound workloads, async doesn’t help because the event loop can’t yield control during computation. Use async for I/O-heavy APIs and consider multiprocessing for CPU-heavy tasks. Most APIs are I/O-bound, so async is usually the right choice.
