How does FastAPI integrate with SQLAlchemy for database connection and ORM?

What you need to know

The rule of thumb: one engine per process, one session per request.

1from typing import Annotated2from fastapi import Depends, FastAPI, HTTPException3from pydantic import BaseModel, ConfigDict4from sqlalchemy import String, select5from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column78engine = create_async_engine("postgresql+asyncpg://...", pool_size=10, pool_pre_ping=True)9SessionLocal = async_sessionmaker(engine, expire_on_commit=False)1011class Base(DeclarativeBase):12    pass1314class Prediction(Base):                      # ORM model: the table15    __tablename__ = "predictions"16    id: Mapped[int] = mapped_column(primary_key=True)17    text: Mapped[str] = mapped_column(String(2000))18    label: Mapped[str] = mapped_column(String(16))19    api_key_hash: Mapped[str] = mapped_column(String(64))2021class PredictionOut(BaseModel):              # Pydantic schema: the API shape22    model_config = ConfigDict(from_attributes=True)23    id: int24    text: str25    label: str2627async def get_db():28    async with SessionLocal() as session:    # one session per request29        yield session                        # closed after the response3031DB = Annotated[AsyncSession, Depends(get_db)]32app = FastAPI()                              # add a lifespan that calls engine.dispose()3334@app.get("/predictions")35async def recent(db: DB, limit: int = 10) -> list[PredictionOut]:36    result = await db.scalars(select(Prediction).order_by(Prediction.id.desc()).limit(limit))37    return result.all()3839@app.get("/predictions/{pid}")40async def read(pid: int, db: DB) -> PredictionOut:41    row = await db.get(Prediction, pid)42    if row is None:43        raise HTTPException(404, "Prediction not found")44    return row

Run against SQLite (sqlite+aiosqlite) with two rows inserted, the real responses are:

GET /predictions    -> [{'id': 2, 'text': 'Great app', 'label': 'positive'},                        {'id': 1, 'text': 'Loved it', 'label': 'positive'}]GET /predictions/7  -> 404

The table has an api_key_hash column, but it is not in the response: PredictionOut lists only three fields, so the return type filtered it out. That is why the two kinds of model are kept separate.

Rules that keep it healthy

• Never share a session between requests. Sessions are not safe for concurrent use; a module-level session corrupts state under load.
• Match the driver to the handler. Async engine and await in async def routes; classic create_engine and Session in plain def routes, which run in the threadpool. A sync query inside async def blocks the event loop.
• Size the pool for your worker count. 4 workers × (10 pool + 5 overflow) = up to 60 connections. Postgres often defaults to 100 connections total, shared with every other service.
• Use Alembic for migrations, not create_all, in real deployments.
• Install sqlalchemy[asyncio] plus a driver such as asyncpg; the async extension needs the greenlet package.

A real-life example

A RAG product logs every question, answer and feedback score to Postgres. It runs 8 FastAPI pods with 4 workers each, and each worker had pool_size=20. That is up to 640 connections against a database allowed 200. At peak, new requests waited for a connection until they timed out, and the dashboards showed "database slow" although the queries themselves took 3 ms.

The fix: pool_size=5, max_overflow=2 per worker (8 × 4 × 7 = 224, then PgBouncer in front to pool further), plus a pool_timeout of 5 seconds so requests fail fast instead of hanging. Pool exhaustion looks exactly like a slow database, so check pool metrics before tuning queries.

Follow-up questions to expect

• "Why expire_on_commit=False?" — After a commit, SQLAlchemy normally expires loaded attributes, so reading them triggers a new query. In async code that lazy query fails outside an await, so we keep the values.
• "Where do you commit?" — In the route or a service function, explicitly. Some teams commit in get_db after yield and roll back on exception; either way, be consistent.
• "What about SQLModel?" — SQLModel, by FastAPI's author, combines the SQLAlchemy table and Pydantic model in one class. It is convenient for simple apps; separate models give more control over what the API exposes.