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
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
UserDBvsUserSchemasplit. - 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
DeclarativeBasesubclasses.
Common pitfalls
table=Truevs pure data model — a class withtable=Truecreates 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) beforeSQLModel.metadata.create_all().
session.refresh(obj)aftercommit()— after a commit, the object's attributes become expired. Callsession.refresh(obj)to reload them, or access attributes within the same session before closing.
Use
select(Model).where(...)instead ofsession.query(Model).filter(...)— theselect()syntax is the SQLAlchemy v2 / SQLModel idiomatic style and works with both sync and async sessions.
Keep
echo=Trueon 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
| Task | Code |
|---|---|
| Define table | class M(SQLModel, table=True): id: Optional[int] = Field(..., primary_key=True) |
| Input model | class MCreate(SQLModel): ... |
| Output model | class MPublic(SQLModel): id: int; ... |
| Create engine | create_engine("sqlite:///db.db") |
| Create tables | SQLModel.metadata.create_all(engine) |
| Session | with Session(engine) as s: |
| Insert | s.add(obj); s.commit() |
| Get by PK | s.get(Model, pk) |
| Select all | s.exec(select(Model)).all() |
| Filter | select(Model).where(Model.col == val) |
| Order | select(Model).order_by(Model.col) |
| Update | obj.field = val; s.add(obj); s.commit() |
| Delete | s.delete(obj); s.commit() |
| Relationship | Relationship(back_populates="other") |
| Async session | AsyncSession + create_async_engine |
| Migrations | Alembic with target_metadata = SQLModel.metadata |