Database Integration with SQLAlchemy or SQLModel in FastAPI

Choosing Between SQLAlchemy and SQLModel

FastAPI integrates cleanly with relational databases through SQLAlchemy (the underlying ORM) or SQLModel (a thin layer that combines SQLAlchemy ORM with Pydantic-style models). Both approaches use the same core ideas: define tables/models, create an engine, open a session per request, and use transactions to persist changes safely.

• SQLAlchemy: most flexible; best when you need advanced mappings, complex relationships, or fine-grained control.
• SQLModel: faster to start; models look like Pydantic models and can double as request/response schemas (still be careful about what you expose).

This chapter shows a production-friendly pattern using SQLAlchemy 2.0 style (recommended) and notes where SQLModel differs.

Step 1: Install Dependencies

Pick a database driver. For local development, SQLite is simplest; for production, PostgreSQL is common.

# SQLite (dev) + SQLAlchemy 2.0 style ORM
pip install sqlalchemy

# PostgreSQL example driver
pip install psycopg[binary]

# Optional: SQLModel alternative
pip install sqlmodel

Step 2: Create the Engine and Session Factory

The engine manages DB connections. A session represents a unit of work. In web apps, you typically create one session per request and close it at the end.

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "sqlite:///./app.db"  # For PostgreSQL: "postgresql+psycopg://user:pass@host:5432/dbname"

# SQLite needs check_same_thread=False when used with FastAPI's threaded execution
connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}

engine = create_engine(DATABASE_URL, echo=False, connect_args=connect_args)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

class Base(DeclarativeBase):
    pass

autocommit=False and autoflush=False make transaction boundaries explicit and reduce surprising writes.

Continue in our app.

You can listen to the audiobook with the screen off, receive a free certificate for this course, and also have access to 5,000 other free online courses.

Or continue reading below...

Download the app

Step 3: Define ORM Models (Tables)

Define tables as Python classes. Keep ORM models focused on persistence concerns; use separate Pydantic schemas for API input/output to avoid leaking internal fields.

# app/models.py
from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column
from .db import Base

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False, index=True)
    description: Mapped[str | None] = mapped_column(String(500), nullable=True)

Step 4: Create Tables

For a beginner-friendly setup, create tables at startup. In real production workflows, you typically use migrations (Alembic), but the pattern below is enough to get a working CRUD feature backed by a database.

# app/main.py
from fastapi import FastAPI
from .db import engine
from .models import Base

app = FastAPI()

@app.on_event("startup")
def on_startup() -> None:
    Base.metadata.create_all(bind=engine)

Step 5: Manage a Session Per Request (Dependency)

Open a session at the start of the request and ensure it closes even if an error occurs. This is the standard pattern to avoid connection leaks.

# app/deps.py
from typing import Generator
from sqlalchemy.orm import Session
from .db import SessionLocal

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 6: Define Pydantic Schemas for Safe I/O

Use separate schemas for create/update vs. read. For responses, enable reading from ORM objects. In Pydantic v2, use ConfigDict(from_attributes=True).

# app/schemas.py
from pydantic import BaseModel, ConfigDict

class ItemCreate(BaseModel):
    name: str
    description: str | None = None

class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None

class ItemRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    description: str | None = None

This prevents accidentally returning fields you did not intend to expose (for example, internal flags, foreign keys, or audit columns).

Step 7: CRUD Functions (Database Layer)

Keep database operations in a dedicated module. This makes endpoints thin and testable.

# app/crud.py
from sqlalchemy.orm import Session
from sqlalchemy import select
from .models import Item
from .schemas import ItemCreate, ItemUpdate

def create_item(db: Session, data: ItemCreate) -> Item:
    item = Item(name=data.name, description=data.description)
    db.add(item)
    db.commit()          # persists transaction
    db.refresh(item)     # loads generated fields like id
    return item

def get_item(db: Session, item_id: int) -> Item | None:
    return db.get(Item, item_id)

def list_items(db: Session, offset: int = 0, limit: int = 100) -> list[Item]:
    stmt = select(Item).offset(offset).limit(limit)
    return list(db.scalars(stmt).all())

def update_item(db: Session, item: Item, data: ItemUpdate) -> Item:
    if data.name is not None:
        item.name = data.name
    if data.description is not None:
        item.description = data.description
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

def delete_item(db: Session, item: Item) -> None:
    db.delete(item)
    db.commit()

Step 8: Map CRUD to API Endpoints

Endpoints call CRUD functions and return response schemas. Notice that the response model is ItemRead, not the ORM model.

# app/routes/items.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from ..deps import get_db
from .. import crud
from ..schemas import ItemCreate, ItemUpdate, ItemRead

router = APIRouter(prefix="/items", tags=["items"])

@router.post("/", response_model=ItemRead, status_code=status.HTTP_201_CREATED)
def create_item(payload: ItemCreate, db: Session = Depends(get_db)):
    item = crud.create_item(db, payload)
    return item

@router.get("/{item_id}", response_model=ItemRead)
def read_item(item_id: int, db: Session = Depends(get_db)):
    item = crud.get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.get("/", response_model=list[ItemRead])
def read_items(offset: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.list_items(db, offset=offset, limit=limit)

@router.patch("/{item_id}", response_model=ItemRead)
def patch_item(item_id: int, payload: ItemUpdate, db: Session = Depends(get_db)):
    item = crud.get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return crud.update_item(db, item, payload)

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def remove_item(item_id: int, db: Session = Depends(get_db)):
    item = crud.get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    crud.delete_item(db, item)
    return None

Include the router in your application:

# app/main.py
from fastapi import FastAPI
from .routes.items import router as items_router
from .db import engine
from .models import Base

app = FastAPI()

@app.on_event("startup")
def on_startup() -> None:
    Base.metadata.create_all(bind=engine)

app.include_router(items_router)

Transaction Handling: Commit/Rollback Patterns

Transactions are where many beginner bugs happen. A few practical rules:

• Commit only when you intend to persist changes. For read-only endpoints, do not commit.
• Rollback on exceptions if you keep the session open after an error. If you always close the session at request end, rollback is still important when you catch exceptions and continue using the session.
• Refresh after commit when you need server-generated values (IDs, defaults, triggers).

Pattern A: Commit inside CRUD functions (simple)

The earlier CRUD functions commit internally. This is easy to follow, but it can be limiting if you need multiple writes to succeed or fail as a single unit.

Pattern B: Commit once in the endpoint (unit-of-work)

For multi-step operations, prefer committing once. CRUD functions only add()/delete() and the endpoint controls the transaction boundary:

# app/crud_uow.py
from sqlalchemy.orm import Session
from .models import Item
from .schemas import ItemCreate

def add_item(db: Session, data: ItemCreate) -> Item:
    item = Item(name=data.name, description=data.description)
    db.add(item)
    return item

# app/routes/items_uow.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ..deps import get_db
from ..schemas import ItemCreate, ItemRead
from ..crud_uow import add_item

router = APIRouter()

@router.post("/items-uow", response_model=ItemRead)
def create_item_uow(payload: ItemCreate, db: Session = Depends(get_db)):
    try:
        item = add_item(db, payload)
        db.commit()
        db.refresh(item)
        return item
    except Exception:
        db.rollback()
        raise

This pattern becomes valuable when you create multiple rows across different tables and want all-or-nothing behavior.

Avoiding Common Pitfalls

Pitfall 1: Returning ORM objects with lazy-loaded relationships

If an ORM object includes relationships configured for lazy loading, Pydantic serialization may trigger additional queries while building the response. This can cause:

• N+1 query problems (one query for the list, then one per row for related data).
• DetachedInstanceError if the session is closed before serialization tries to access a relationship.

Solutions:

• Prefer explicit response schemas that only include fields you know are loaded.
• Eager load relationships when you need them (e.g., selectinload or joinedload).
• Do not rely on implicit lazy loading during response serialization.

# Example eager loading (if Item had a relationship like Item.owner)
from sqlalchemy import select
from sqlalchemy.orm import selectinload

stmt = select(Item).options(selectinload(Item.owner))
items = db.scalars(stmt).all()

Pitfall 2: Forgetting to refresh after commit

After commit(), values generated by the database (like auto-increment IDs) may not be present until you call refresh() or access them in a way that triggers a refresh. Calling db.refresh(obj) makes the behavior explicit and predictable.

Pitfall 3: Using a session across threads/tasks

A SQLAlchemy session is not designed to be shared across concurrent tasks. Keep the session scoped to the request and pass it down to functions. If you start background work, create a new session inside that background context.

Pitfall 4: Accidental partial updates

For PATCH-like updates, distinguish between “field not provided” and “field provided as null”. The ItemUpdate schema above uses optional fields; your update logic checks is not None to avoid overwriting values unintentionally.

Fully Working CRUD Feature: File Layout

A minimal, working structure for the CRUD feature shown above:

app/
  __init__.py
  main.py
  db.py
  deps.py
  models.py
  schemas.py
  crud.py
  routes/
    __init__.py
    items.py

SQLModel Variant (Conceptual Mapping)

If you choose SQLModel, the flow is the same: create an engine, create tables, use a session dependency, and implement CRUD. The main difference is that models can be declared with SQLModel and can optionally serve as response models. Even then, it is often safer to keep separate read/write schemas for public API boundaries.

# SQLModel sketch (not a full replacement of the SQLAlchemy code above)
from sqlmodel import SQLModel, Field

class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    description: str | None = None

Whether you use SQLAlchemy or SQLModel, the key production-ready practices remain: one session per request, explicit transaction boundaries, explicit loading of data needed for responses, and response schemas that control what leaves your API.