Async database access in FastAPI: lessons from production

The async database access problem in production

You've built a FastAPI endpoint that feels snappy in development, but the moment you scale it to handle real traffic, you hit a wall: database queries block your async workers, defeating the purpose of async altogether. We've seen this play out across e-commerce platforms handling thousands of concurrent checkout requests and SaaS applications managing real-time data pipelines. Async database access in FastAPI: lessons from production shows us that the real bottleneck isn't always your code—it's the mismatch between async routes and synchronous database drivers.

Why async routes need async drivers

When you write an async endpoint in FastAPI, you're making a contract with the framework: "This coroutine will yield control back to the event loop so other requests can be processed." The problem is that most traditional database drivers—including the synchronous variant of SQLAlchemy and psycopg2—are blocking. They don't yield; they wait.

Here's what happens when you use a blocking driver in an async context:

from fastapi import FastAPI
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker

app = FastAPI()
engine = create_engine("postgresql://user:password@localhost/db")
SessionLocal = sessionmaker(bind=engine)

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    db = SessionLocal()
    # This blocks the event loop for the entire query duration
    order = db.query(Order).filter(Order.id == order_id).first()
    db.close()
    return order

When the query runs, your async endpoint freezes the entire event loop. Other requests queue up. Your concurrency advantage evaporates. In a system handling 100 concurrent users, you might end up with the throughput of a synchronous application—or worse.

The solution is straightforward: use an async-aware database driver. For PostgreSQL, that's asyncpg. For broader compatibility, SQLAlchemy's async engine provides a clean abstraction layer.

Using asyncpg or SQLAlchemy async engine

AsyncPG is a purpose-built async driver for PostgreSQL. It's fast, lightweight, and plays nicely with asyncio:

import asyncpg
from fastapi import FastAPI

app = FastAPI()

async def get_db_pool():
    return await asyncpg.create_pool(
        user="user",
        password="password",
        database="db",
        host="localhost",
        min_size=10,
        max_size=20
    )

pool = None

@app.on_event("startup")
async def startup():
    global pool
    pool = await get_db_pool()

@app.on_event("shutdown")
async def shutdown():
    await pool.close()

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    async with pool.acquire() as connection:
        order = await connection.fetchrow(
            "SELECT * FROM orders WHERE id = $1",
            order_id
        )
    return order

This approach works, but many teams prefer the ORM abstraction layer. SQLAlchemy 2.0+ includes an async engine that wraps asyncpg while maintaining familiar ORM patterns:

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker
)
from sqlalchemy import select

app = FastAPI()

# Use the async dialect
engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/db",
    echo=False,
    pool_size=20,
    max_overflow=0
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

async def get_session():
    async with async_session() as session:
        yield session

@app.get("/orders/{order_id}")
async def get_order(order_id: int, session: AsyncSession = Depends(get_session)):
    stmt = select(Order).where(Order.id == order_id)
    result = await session.execute(stmt)
    order = result.scalars().first()
    return order

The async engine ensures that all I/O operations genuinely yield to the event loop. Your endpoints stay non-blocking, and you retain the benefits of ORM patterns: relationships, lazy loading (with care), query composition, and migrations.

Do not call sync drivers in async def

This is the most common production mistake we've seen: developers assume that wrapping a sync driver call in await or asyncio.run() will solve the problem. It won't.

Consider this antipattern:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import asyncio

sync_engine = create_engine("postgresql://user:password@localhost/db")
SessionLocal = sessionmaker(bind=sync_engine)

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    # DON'T DO THIS
    order = await asyncio.to_thread(
        lambda: SessionLocal().query(Order).filter(Order.id == order_id).first()
    )
    return order

While asyncio.to_thread() pushes the blocking call to a thread pool, you've now:

• Added thread pool overhead and context switching cost
• Lost the scaling benefits of async (thread pools are finite; event loops handle thousands of concurrent tasks)
• Created potential connection pool contention if multiple threads compete for the same database connections

A variant we've encountered: using asyncio.run() inside an async function. This is even worse—it spawns a new event loop inside an existing one, which FastAPI's lifespan management doesn't handle gracefully:

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    # NEVER DO THIS
    order = asyncio.run(fetch_order_sync(order_id))
    return order

The fix is simple: migrate to an async driver. The effort is usually under a day for most applications, and the payoff in throughput and resource utilization is substantial.

Or keep routes sync — FastAPI runs them in a thread pool

Not every endpoint needs to be async. FastAPI has a pragmatic fallback: you can write synchronous route handlers, and the framework will run them in a thread pool, freeing up the event loop for other async work.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

sync_engine = create_engine("postgresql://user:password@localhost/db")
SessionLocal = sessionmaker(bind=sync_engine)

@app.get("/orders/{order_id}")  # Note: no 'async' keyword
def get_order(order_id: int):
    db = SessionLocal()
    order = db.query(Order).filter(Order.id == order_id).first()
    db.close()
    return order

This is a valid strategy if:

• Your endpoints have few external I/O operations (database queries dominate)
• You're in the early stages of a project and need to move fast
• You have existing synchronous code that's hard to refactor
• You've measured and confirmed that thread pool latency is acceptable for your use case

In a CRM system we worked with, roughly 60% of endpoints were converted to async (complex multi-query flows), while 40% remained sync (simple CRUD operations). The mixed approach worked well because the thread pool handled the synchronous work without competing for event loop cycles.

The trade-off: you lose the concurrency advantage for those endpoints. With 20 worker threads in the pool, you can handle maybe 20 concurrent sync requests before queuing. With async, you handle thousands. Choose sync when the simplicity wins and the load justifies it.

Measure before going all-in on async

Async is not a silver bullet. We've seen teams spend weeks converting to async drivers only to find that their bottleneck was elsewhere: slow queries, missing indexes, or external API calls. Before committing to a full refactor, measure.

Use tools like Apache Bench, wrk, or Locust to profile your current setup:

import time
from fastapi import FastAPI

app = FastAPI()

@app.middleware("http")
async def log_timing(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Process-Time"] = str(elapsed)
    return response

Run load tests against your sync routes and measure:

• Throughput (requests per second)
• Latency percentiles (p50, p95, p99)
• CPU and memory usage
• Database connection pool saturation

Then profile your database layer directly. A query that takes 100ms will limit you to 10 requests per second per worker, regardless of async or sync. If your slow queries are the real bottleneck, optimize those first.

Once you've validated that database latency is your constraint and that concurrent requests are piling up, then migrate to async. You'll see the gain clearly.

Common mistakes

• Mixing sync and async drivers: Creating two engine instances (one sync, one async) and using them in the same codebase leads to connection pool confusion. Use one async engine across your application.
• Forgetting to configure connection pooling: With async, you can handle more concurrent requests, but each one needs a database connection. Under-provisioning the pool will starve requests. A rule of thumb: pool_size = number_of_workers + buffer.
• Awaiting without actually yielding: Libraries that claim to be async but block internally (e.g., calling time.sleep() instead of asyncio.sleep()) will still freeze your event loop. Audit third-party libraries carefully.
• Holding connections across multiple I/O operations without care: Async allows you to spawn many concurrent requests, but if each holds a database connection for its entire lifetime, you'll exhaust the pool. Use context managers and yield connections as soon as possible.

Key takeaways

• Async routes paired with blocking database drivers give you the worst of both worlds: the complexity of async without the concurrency benefit.
• Use asyncpg for raw SQL or SQLAlchemy's async engine for ORM-style code. Both genuinely yield to the event loop.
• Sync routes in FastAPI are a pragmatic alternative if your endpoints are I/O-light or you're not ready to refactor. The framework handles them efficiently via a thread pool.
• Measure your baseline throughput, latency, and resource usage before investing in async. The real bottleneck might be query performance, not concurrency.
• If you do go async, pay attention to connection pooling and ensure you're not holding connections longer than necessary.