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#

pip install sqlmodel

Output: (none — exits 0 on success)

SQLModel pins SQLAlchemy v2 and Pydantic v2 as dependencies.

Quick example#

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:

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#

[!WARNING] 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.

[!WARNING] 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().

[!WARNING] 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.

[!TIP] 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.

[!TIP] 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.

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#

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:

1 Deadpond
2 Rusty-Man
3 Spider-Boy

# READ — filtered
with Session(engine) as session:
    young = session.exec(select(Hero).where(Hero.age < 30)).all()
    print(young)

Output:

[Hero(id=3, name='Spider-Boy', secret_name='Pedro Parqueador', age=16)]

# 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:

id=1 name='Deadpond' secret_name='Dive Wilson' age=35

# DELETE
with Session(engine) as session:
    hero = session.get(Hero, 3)
    session.delete(hero)
    session.commit()
    print(f"Deleted: {hero.name}")

Output:

Deleted: Spider-Boy

Filtering and ordering#

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

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:

Rusty-Man 48

# 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#

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:

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.

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.

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

Output: (none — exits 0 on success)

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.

pip install alembic
alembic init alembic

Output: (none — exits 0 on success)

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

from sqlmodel import SQLModel
import app.models   # import all your models so they register with SQLModel.metadata
target_metadata = SQLModel.metadata

alembic revision --autogenerate -m "add hero table"
alembic upgrade head

Output: (none — exits 0 on success)

Quick reference#