cheat sheet

SQLModel

Define database models with Python type hints using SQLModel. Covers table creation, CRUD, relationships, FastAPI integration, Pydantic validation, and async sessions.

SQLModel — Typed ORM for FastAPI

What it is

SQLModel is a Python ORM library that combines SQLAlchemy (database engine) and Pydantic v2 (data validation) behind a single model class. You define one SQLModel class that serves as both the database table schema and the Pydantic validation model — no duplication between ORM models and API schemas. It was created by the same author as FastAPI and is designed to slot directly into FastAPI request/response handling.

Install

bash
pip install sqlmodel

Output: (none — exits 0 on success)

SQLModel pins SQLAlchemy v2 and Pydantic v2 as dependencies.

Quick example

python
from sqlmodel import Field, Session, SQLModel, create_engine, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None

engine = create_engine("sqlite:///heroes.db", echo=False)
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    hero = Hero(name="Spider-Boy", secret_name="Pedro Parqueador", age=16)
    session.add(hero)
    session.commit()
    session.refresh(hero)
    print(hero)

Output:

text
id=1 name='Spider-Boy' secret_name='Pedro Parqueador' age=16

When / why to use it

  • FastAPI projects where you want a single model for both DB schema and request/response validation.
  • Replacing the SQLAlchemy + Pydantic duplication: one class, no UserDB vs UserSchema split.
  • Typed database access with .model_validate(), .model_dump(), and IDE autocomplete on results.
  • SQLite, PostgreSQL, MySQL — any database SQLAlchemy supports.
  • Gradual migration from raw SQLAlchemy: SQLModel models are fully compatible SQLAlchemy DeclarativeBase subclasses.

Common pitfalls

table=True vs pure data model — a class with table=True creates a DB table. Without it, the class is a pure Pydantic model used for validation only (e.g. API request bodies). Mix both patterns: HeroCreate(SQLModel) for input, Hero(SQLModel, table=True) for the table, HeroPublic(SQLModel) for output.

Circular import with Relationship — relationship fields that reference another model by string name (back_populates="heroes") require both model files to be imported before the session is used. Import all models in one place (e.g. models/__init__.py) before SQLModel.metadata.create_all().

session.refresh(obj) after commit() — after a commit, the object's attributes become expired. Call session.refresh(obj) to reload them, or access attributes within the same session before closing.

Use select(Model).where(...) instead of session.query(Model).filter(...) — the select() syntax is the SQLAlchemy v2 / SQLModel idiomatic style and works with both sync and async sessions.

Keep echo=True on the engine during development: create_engine(url, echo=True). SQLAlchemy prints every SQL statement to stdout, making it easy to spot N+1 queries.

Data model patterns — input, table, output

The recommended pattern separates three concerns into three classes that share field definitions via inheritance.

python
from sqlmodel import Field, SQLModel

# 1. Input (no ID — client doesn't provide it)
class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: int | None = None

# 2. Table (inherits fields, adds DB-specific columns)
class Hero(HeroCreate, table=True):
    id: int | None = Field(default=None, primary_key=True)

# 3. Output (expose only safe fields)
class HeroPublic(SQLModel):
    id: int
    name: str
    age: int | None = None

CRUD operations

python
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Optional

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

engine = create_engine("sqlite:///heroes.db")
SQLModel.metadata.create_all(engine)

# CREATE
with Session(engine) as session:
    heroes = [
        Hero(name="Deadpond",   secret_name="Dive Wilson"),
        Hero(name="Rusty-Man",  secret_name="Tommy Sharp",  age=48),
        Hero(name="Spider-Boy", secret_name="Pedro Parqueador", age=16),
    ]
    session.add_all(heroes)
    session.commit()

# READ — all
with Session(engine) as session:
    result = session.exec(select(Hero))
    for hero in result:
        print(hero.id, hero.name)

Output:

text
1 Deadpond
2 Rusty-Man
3 Spider-Boy
python
# READ — filtered
with Session(engine) as session:
    young = session.exec(select(Hero).where(Hero.age < 30)).all()
    print(young)

Output:

text
[Hero(id=3, name='Spider-Boy', secret_name='Pedro Parqueador', age=16)]
python
# UPDATE
with Session(engine) as session:
    hero = session.get(Hero, 1)   # get by primary key
    hero.age = 35
    session.add(hero)
    session.commit()
    session.refresh(hero)
    print(hero)

Output:

text
id=1 name='Deadpond' secret_name='Dive Wilson' age=35
python
# DELETE
with Session(engine) as session:
    hero = session.get(Hero, 3)
    session.delete(hero)
    session.commit()
    print(f"Deleted: {hero.name}")

Output:

text
Deleted: Spider-Boy

Filtering and ordering

The select() + .where() + .order_by() chain maps directly to SQL.

python
from sqlmodel import Session, select, col

with Session(engine) as session:
    # Multiple conditions (AND)
    results = session.exec(
        select(Hero)
        .where(Hero.age >= 18)
        .where(Hero.age <= 50)
        .order_by(Hero.age)
        .limit(10)
        .offset(0)
    ).all()
    for h in results:
        print(h.name, h.age)

Output:

text
Rusty-Man 48
python
# LIKE / ILIKE
results = session.exec(
    select(Hero).where(col(Hero.name).contains("Man"))
).all()

# IN
results = session.exec(
    select(Hero).where(col(Hero.id).in_([1, 2]))
).all()

Relationships — one-to-many

python
from typing import Optional, List
from sqlmodel import Field, Session, SQLModel, create_engine, Relationship, select

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    headquarters: str
    heroes: List["Hero"] = Relationship(back_populates="team")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    team_id: Optional[int] = Field(default=None, foreign_key="team.id")
    team: Optional[Team] = Relationship(back_populates="heroes")

engine = create_engine("sqlite:///teams.db")
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    team = Team(name="Preventers", headquarters="Sharp Tower")
    session.add(team)
    session.commit()
    session.refresh(team)

    hero = Hero(name="Rusty-Man", secret_name="Tommy Sharp", team_id=team.id)
    session.add(hero)
    session.commit()
    session.refresh(hero)

    # Load with relationship
    team_db = session.get(Team, team.id)
    print(team_db.name, [h.name for h in team_db.heroes])

Output:

text
Preventers ['Rusty-Man']

FastAPI integration

SQLModel's dual nature means one model class works as both FastAPI's Pydantic body and the SQLAlchemy table row.

python
from fastapi import FastAPI, HTTPException, Depends
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Optional, List

sqlite_url = "sqlite:///./app.db"
engine = create_engine(sqlite_url)

class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None

class Hero(HeroCreate, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroPublic(SQLModel):
    id: int
    name: str
    age: Optional[int] = None

def get_session():
    with Session(engine) as session:
        yield session

app = FastAPI()

@app.on_event("startup")
def on_startup():
    SQLModel.metadata.create_all(engine)

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: Session = Depends(get_session)):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(offset: int = 0, limit: int = 10, session: Session = Depends(get_session)):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: Session = Depends(get_session)):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroCreate, session: Session = Depends(get_session)):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: Session = Depends(get_session)):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

Async sessions

For async FastAPI (with asyncio), use AsyncSession and create_async_engine.

bash
pip install aiosqlite    # async SQLite driver
pip install asyncpg      # async PostgreSQL driver

Output: (none — exits 0 on success)

python
from sqlmodel import SQLModel, Field, select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from typing import Optional

DATABASE_URL = "sqlite+aiosqlite:///./app.db"
async_engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str

async def create_db():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

async def get_heroes() -> list[Hero]:
    async with AsyncSessionLocal() as session:
        result = await session.exec(select(Hero))
        return result.all()

Migrations with Alembic

SQLModel is compatible with Alembic for schema migrations.

bash
pip install alembic
alembic init alembic

Output: (none — exits 0 on success)

In alembic/env.py, point target_metadata at SQLModel's metadata:

python
from sqlmodel import SQLModel
import app.models   # import all your models so they register with SQLModel.metadata
target_metadata = SQLModel.metadata
bash
alembic revision --autogenerate -m "add hero table"
alembic upgrade head

Output: (none — exits 0 on success)

Quick reference

TaskCode
Define tableclass M(SQLModel, table=True): id: Optional[int] = Field(..., primary_key=True)
Input modelclass MCreate(SQLModel): ...
Output modelclass MPublic(SQLModel): id: int; ...
Create enginecreate_engine("sqlite:///db.db")
Create tablesSQLModel.metadata.create_all(engine)
Sessionwith Session(engine) as s:
Inserts.add(obj); s.commit()
Get by PKs.get(Model, pk)
Select alls.exec(select(Model)).all()
Filterselect(Model).where(Model.col == val)
Orderselect(Model).order_by(Model.col)
Updateobj.field = val; s.add(obj); s.commit()
Deletes.delete(obj); s.commit()
RelationshipRelationship(back_populates="other")
Async sessionAsyncSession + create_async_engine
MigrationsAlembic with target_metadata = SQLModel.metadata