Async Database and Concurrency (Optional)

Introduction

FastAPI supports async def routes, but sync SQLAlchemy blocks the event loop if you call it inside async routes without care. Async SQLAlchemy with AsyncSession fits I/O-heavy APIs. This chapter is optional—the main track uses sync SQLAlchemy until you measure a real need.

Prerequisites

SQLAlchemy With FastAPI
Basic async/await concepts
pip install sqlalchemy[asyncio] aiomysql (MySQL) or aiosqlite (SQLite dev)

When to Use Async DB

Situation	Recommendation
CRUD API, moderate traffic	Sync SQLAlchemy + `def` routes (simpler)
Many concurrent DB waits, async drivers	`AsyncSession` + `async def`
CPU-heavy work in route	Thread pool or worker queue—not raw async

Async helps when the app waits on I/O—not when CPU-bound Python runs hot.

Async Engine and Session

app/db/async_session.py:

python

MySQL URL example:

text

mysql+aiomysql://user:pass@localhost:3306/app?charset=utf8mb4

Code explanation:

Driver must be async-capable (aiomysql, asyncpg, aiosqlite)
Use async with session context

Async Route Query

python

from sqlalchemy import select
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
 
@app.get("/items/", response_model=list[ItemRead])
async def list_items(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(Item).limit(20))
    items = result.scalars().all()
    return items

Create:

python

@app.post("/items/", response_model=ItemRead, status_code=201)
async def create_item(payload: ItemCreate, db: AsyncSession = Depends(get_async_db)):
    item = Item(name=payload.name, price=payload.price)
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

Sync ORM in Async Route (Avoid)

Bad:

python

@app.get("/bad")
async def bad_route(db: Session = Depends(get_db)):
    return db.query(Item).all()

Blocks event loop under load.

Better—use sync route:

python

@app.get("/good")
def good_route(db: Session = Depends(get_db)):
    return db.scalars(select(Item)).all()

Or run sync code in thread pool:

python

from fastapi.concurrency import run_in_threadpool
 
@app.get("/offload")
async def offload(db: Session = Depends(get_db)):
    def query():
        return db.scalars(select(Item)).all()
    return await run_in_threadpool(query)

Lifespan With Async Engine

python

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()

Dispose async engine on shutdown.

Alembic With Async

Alembic migrations typically run sync even when app is async—use sync URL in alembic/env.py or Alembic async template. Many teams keep alembic upgrade sync in deploy scripts.

Async Tests

python

@pytest.mark.asyncio
async def test_list_items(async_client):
    response = await async_client.get("/items/")
    assert response.status_code == 200

Override get_async_db with test session fixture—mirror testing chapter.

FAQ

Mix sync and async routes?

Allowed—Uvicorn runs async app; sync routes use thread pool internally.

AsyncSession greenlet errors?

Use await on all DB calls; check SQLAlchemy 2.0 async docs.

Same models for sync and async?

Yes—Base models shared; two session factories if migrating gradually.

Performance magic?

Profile first—async adds complexity; sync + workers scales many APIs fine.

FastAPI + Django ORM async?

Unusual—stick to SQLAlchemy in this track.

Skip this chapter?

Yes—complete projects with sync SQLAlchemy unless requirements demand async.