FastAPI Async Database Access

A database query spends most of its time waiting for the database server to answer. A synchronous route freezes its worker during that wait. An async route hands control back to the event loop, which serves other requests while the query runs. Think of a waiter who places an order in the kitchen and keeps serving other tables instead of standing at the pass until the dish is ready.

The Waiting Time Picture

Sync route:    [send query][........ waiting ........][use result]
                           worker blocked

Async route:   [send query]  ──▶ event loop serves other requests ──▶ [use result]
                                  worker stays busy

Busy APIs gain the most, because many requests overlap during the waits.

Installing the Pieces

SQLAlchemy needs an async driver for your database.

pip install sqlalchemy aiosqlite     # SQLite
pip install sqlalchemy asyncpg       # PostgreSQL

Creating the Async Engine and Session Factory

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

DATABASE_URL = "sqlite+aiosqlite:///./app.db"
# PostgreSQL: "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

The driver name appears in the URL after the plus sign. The setting expire_on_commit=False keeps objects readable after a commit, which suits async code.

The Session Dependency

from fastapi import Depends

async def get_db():
    async with SessionLocal() as session:
        yield session

Every request receives its own session. The async with block closes it when the request ends.

Creating Tables at Startup

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

app = FastAPI(lifespan=lifespan)

The run_sync call lets the table-creation helper run inside an async connection. Production projects replace this step with Alembic migrations.

Reading Data

from sqlalchemy import select

@app.get("/items")
async def list_items(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Item))
    return result.scalars().all()

@app.get("/items/{item_id}")
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    item = await db.get(Item, item_id)
    return item

Every database call carries the await keyword. Missing it returns a coroutine object instead of data.

Writing Data

@app.post("/items", status_code=201)
async def create_item(data: ItemCreate, db: AsyncSession = Depends(get_db)):
    item = Item(**data.model_dump())
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

db.add(item)        ──▶  stages the new row in memory (no await needed)
await db.commit()   ──▶  writes the row to the database
await db.refresh()  ──▶  reloads generated values such as the new id

Loading Related Data

Async sessions refuse lazy loading. Touching author.books without preparation raises a MissingGreenlet error. Request the related rows inside the query.

from sqlalchemy.orm import selectinload

stmt = select(Author).options(selectinload(Author.books))
result = await db.execute(stmt)
authors = result.scalars().all()

Choosing Sync or Async

• Pick async when the app handles many simultaneous requests or calls other async services.
• Pick sync for small projects and simple scripts; the code stays shorter.
• Avoid mixing styles. A sync database call inside an async def route blocks the entire event loop.

Common Mistakes

Using a Sync Driver

The URL sqlite:/// belongs to the sync engine. Async engines need sqlite+aiosqlite:///.

Forgetting await

The route returns a strange object or a warning about a coroutine that never ran. Add await before every session call that talks to the database.

Sharing One Session

A global session breaks when two requests overlap. Create one session per request through the dependency.

A Small Timing Experiment

A quick experiment makes the benefit visible. Imagine a route that waits one second for the database. Ten visitors arrive at the same moment.

Sync worker, one process:      10 requests x 1 second  =  about 10 seconds for the last visitor
Async worker, one process:     10 requests overlap     =  about 1 second for every visitor

The async worker starts all ten queries and collects the answers as they arrive. The sync worker finishes one query before it touches the next. Real databases answer faster than one second, yet the same pattern holds at larger scale: a thousand overlapping requests make the gap far bigger.

Measure before you convert an existing project. Apps with light traffic gain little from async access, and the extra keywords add reading effort for new team members.

Key Points

• Async access frees the worker while the database answers.
• Use create_async_engine and AsyncSession with an async driver.
• Add await before execute, commit, refresh, and get.
• Load related rows with selectinload because lazy loading fails.
• Give each request its own session through a dependency.