SQLModel combines SQLAlchemy and Pydantic for type-safe ORM models. pip install sqlmodel. Model: from sqlmodel import SQLModel, Field; class User(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True); name: str. Engine: from sqlmodel import create_engine; engine = create_engine("sqlite:///db.sqlite3"). Create tables: SQLModel.metadata.create_all(engine). Session: from sqlmodel import Session; with Session(engine) as s: s.add(user); s.commit(). Select: from sqlmodel import select; s.exec(select(User).where(User.name == "Alice")).first(). All: s.exec(select(User)).all(). Get by PK: s.get(User, 1). Update: user.name = "Bob"; s.add(user); s.commit(); s.refresh(user). Delete: s.delete(user); s.commit(). Field options: Field(default=None, primary_key=True), Field(index=True), Field(unique=True), Field(nullable=False). Relationship: from sqlmodel import Relationship; class Hero(SQLModel, table=True): team_id: int | None = Field(default=None, foreign_key="team.id"); team: "Team" = Relationship(back_populates="heroes"). Validator: Pydantic validators work on SQLModel. Async: from sqlmodel.ext.asyncio.session import AsyncSession; from sqlalchemy.ext.asyncio import create_async_engine. engine = create_async_engine("sqlite+aiosqlite:///db.sqlite3"). FastAPI: inject SessionDep = Annotated[Session, Depends(get_session)]. Claude Code generates SQLModel CRUD services, FastAPI database dependencies, and async session patterns.
CLAUDE.md for SQLModel
## SQLModel Stack
- Version: sqlmodel >= 0.0.19 | pip install sqlmodel sqlalchemy
- Model: class User(SQLModel, table=True): id: int|None = Field(primary_key=True)
- Engine: create_engine("sqlite:///db.sqlite3") | postgres: postgresql://...
- CRUD: with Session(engine) as s: s.add(obj); s.commit(); s.refresh(obj)
- Query: s.exec(select(Model).where(Model.col == val)).all()
- Async: create_async_engine + AsyncSession for FastAPI async endpointsSQLModel CRUD Pipeline
# app/db_sqlmodel.py — SQLModel models, CRUD, relationships, and FastAPI integration
from __future__ import annotations
from datetime import datetime
from typing import Optional, Generator, AsyncGenerator
from sqlalchemy import event
from sqlmodel import (
Field,
Relationship,
Session,
SQLModel,
create_engine,
select,
)
# ─────────────────────────────────────────────────────────────────────────────
# 1. Models
# ─────────────────────────────────────────────────────────────────────────────
class TeamBase(SQLModel):
name: str = Field(index=True)
division: str | None = None
class Team(TeamBase, table=True):
id: int | None = Field(default=None, primary_key=True)
heroes: list["Hero"] = Relationship(back_populates="team")
class TeamCreate(TeamBase):
pass
class TeamRead(TeamBase):
id: int
class HeroBase(SQLModel):
name: str = Field(index=True)
secret_name: str
age: int | None = None
active: bool = Field(default=True)
class Hero(HeroBase, table=True):
id: int | None = Field(default=None, primary_key=True)
team_id: int | None = Field(default=None, foreign_key="team.id", index=True)
created_at: datetime = Field(default_factory=datetime.utcnow)
team: Optional[Team] = Relationship(back_populates="heroes")
class HeroCreate(HeroBase):
team_id: int | None = None
class HeroRead(HeroBase):
id: int
team_id: int | None
created_at: datetime
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
active: bool | None = None
team_id: int | None = None
# ─────────────────────────────────────────────────────────────────────────────
# 2. Engine and session factory
# ─────────────────────────────────────────────────────────────────────────────
def make_engine(url: str = "sqlite:///./heroes.db", echo: bool = False):
"""
Create a SQLAlchemy engine.
url: SQLAlchemy connection string.
SQLite: "sqlite:///./app.db"
PostgreSQL: "postgresql+psycopg2://user:pass@host/db"
MySQL: "mysql+pymysql://user:pass@host/db"
"""
engine = create_engine(url, echo=echo)
# Enable WAL mode for SQLite
if url.startswith("sqlite"):
@event.listens_for(engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
return engine
def init_db(engine) -> None:
"""Create all tables."""
SQLModel.metadata.create_all(engine)
def get_session(engine) -> Generator[Session, None, None]:
"""
FastAPI dependency: yields a Session for the request lifecycle.
Usage:
@app.get("/heroes")
def list_heroes(session: Session = Depends(get_session(engine))):
...
"""
with Session(engine) as session:
yield session
# ─────────────────────────────────────────────────────────────────────────────
# 3. CRUD operations
# ─────────────────────────────────────────────────────────────────────────────
class HeroService:
"""
Service layer for Hero CRUD using SQLModel.
Usage:
svc = HeroService(session)
hero = svc.create(HeroCreate(name="Spider-Man", secret_name="Peter Parker"))
heroes = svc.list_active(limit=10)
"""
def __init__(self, session: Session) -> None:
self._s = session
# --- Create ---
def create(self, data: HeroCreate) -> Hero:
hero = Hero.model_validate(data)
self._s.add(hero)
self._s.commit()
self._s.refresh(hero)
return hero
def create_many(self, items: list[HeroCreate]) -> list[Hero]:
heroes = [Hero.model_validate(d) for d in items]
self._s.add_all(heroes)
self._s.commit()
for h in heroes:
self._s.refresh(h)
return heroes
# --- Read ---
def get(self, hero_id: int) -> Hero | None:
return self._s.get(Hero, hero_id)
def get_or_404(self, hero_id: int) -> Hero:
hero = self.get(hero_id)
if hero is None:
raise KeyError(f"Hero {hero_id} not found")
return hero
def list_all(self, limit: int = 100, offset: int = 0) -> list[Hero]:
return self._s.exec(
select(Hero).offset(offset).limit(limit)
).all()
def list_active(self, limit: int = 100) -> list[Hero]:
return self._s.exec(
select(Hero).where(Hero.active == True).limit(limit)
).all()
def find_by_name(self, name: str) -> Hero | None:
return self._s.exec(
select(Hero).where(Hero.name == name)
).first()
def by_team(self, team_id: int) -> list[Hero]:
return self._s.exec(
select(Hero).where(Hero.team_id == team_id)
).all()
def count(self, active_only: bool = False) -> int:
from sqlmodel import func
stmt = select(func.count()).select_from(Hero)
if active_only:
stmt = stmt.where(Hero.active == True)
return self._s.exec(stmt).one()
# --- Update ---
def update(self, hero_id: int, data: HeroUpdate) -> Hero:
hero = self.get_or_404(hero_id)
update_data = data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(hero, field, value)
self._s.add(hero)
self._s.commit()
self._s.refresh(hero)
return hero
# --- Delete ---
def delete(self, hero_id: int) -> bool:
hero = self.get(hero_id)
if hero is None:
return False
self._s.delete(hero)
self._s.commit()
return True
class TeamService:
def __init__(self, session: Session) -> None:
self._s = session
def create(self, data: TeamCreate) -> Team:
team = Team.model_validate(data)
self._s.add(team)
self._s.commit()
self._s.refresh(team)
return team
def get(self, team_id: int) -> Team | None:
return self._s.get(Team, team_id)
def list_all(self) -> list[Team]:
return self._s.exec(select(Team)).all()
# ─────────────────────────────────────────────────────────────────────────────
# 4. FastAPI integration
# ─────────────────────────────────────────────────────────────────────────────
FASTAPI_EXAMPLE = '''
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, create_engine, SQLModel
from app.db_sqlmodel import (
Hero, HeroCreate, HeroRead, HeroUpdate,
HeroService, init_db, get_session,
)
engine = create_engine("sqlite:///./heroes.db")
init_db(engine)
app = FastAPI()
SessionDep = Annotated[Session, Depends(lambda: get_session(engine))]
@app.post("/heroes/", response_model=HeroRead)
def create_hero(hero: HeroCreate, session: SessionDep):
return HeroService(session).create(hero)
@app.get("/heroes/", response_model=list[HeroRead])
def list_heroes(session: SessionDep, limit: int = 20, offset: int = 0):
return HeroService(session).list_all(limit=limit, offset=offset)
@app.get("/heroes/{hero_id}", response_model=HeroRead)
def get_hero(hero_id: int, session: SessionDep):
hero = HeroService(session).get(hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroRead)
def update_hero(hero_id: int, data: HeroUpdate, session: SessionDep):
return HeroService(session).update(hero_id, data)
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
if not HeroService(session).delete(hero_id):
raise HTTPException(status_code=404, detail="Hero not found")
return {"ok": True}
'''
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
engine = make_engine("sqlite:///./demo_heroes.db")
init_db(engine)
with Session(engine) as s:
team_svc = TeamService(s)
hero_svc = HeroService(s)
print("=== Create teams ===")
avengers = team_svc.create(TeamCreate(name="Avengers", division="Alpha"))
xmen = team_svc.create(TeamCreate(name="X-Men", division="Beta"))
print(f" {avengers.id}: {avengers.name}")
print("\n=== Create heroes ===")
hero_svc.create_many([
HeroCreate(name="Spider-Man", secret_name="Peter Parker", age=28, team_id=avengers.id),
HeroCreate(name="Iron Man", secret_name="Tony Stark", age=45, team_id=avengers.id),
HeroCreate(name="Wolverine", secret_name="Logan", age=200,team_id=xmen.id),
HeroCreate(name="Storm", secret_name="Ororo Munroe", age=35, team_id=xmen.id),
])
print("\n=== List all ===")
for h in hero_svc.list_all():
print(f" [{h.id}] {h.name} (team_id={h.team_id})")
print("\n=== Find by name ===")
sm = hero_svc.find_by_name("Spider-Man")
print(f" Found: {sm.name}, age={sm.age}")
print("\n=== Update ===")
hero_svc.update(sm.id, HeroUpdate(age=29))
sm_updated = hero_svc.get(sm.id)
print(f" Updated age: {sm_updated.age}")
print("\n=== By team ===")
avengers_heroes = hero_svc.by_team(avengers.id)
print(f" Avengers: {[h.name for h in avengers_heroes]}")
print(f"\n=== Count: {hero_svc.count()} heroes ===")For the SQLAlchemy + Pydantic separately alternative — using SQLAlchemy Core/ORM for models and Pydantic separately requires maintaining two parallel class hierarchies (one for DB, one for API validation); SQLModel unifies them with table=True models that are both SQLAlchemy models and Pydantic schemas, eliminating the need for .from_orm() or manual field duplication. For the tortoise-orm alternative — tortoise-orm uses async with transactions and is built for async-first designs with Model.create() / Model.filter(); SQLModel wraps SQLAlchemy giving access to the full SQLAlchemy ecosystem (Alembic migrations, Core expressions, connection pooling) — SQLModel is the recommended ORM when building FastAPI applications where Alembic migrations and Pydantic validation are important. The Claude Skills 360 bundle includes SQLModel skill sets covering Hero/Team models with SQLModel base classes, Field() with primary_key/index/foreign_key/default, HeroCreate/HeroRead/HeroUpdate split schemas, make_engine()/init_db()/get_session() setup, HeroService/TeamService CRUD classes, FastAPI Depends(get_session) dependency injection, select().where().offset().limit() queries, and async engine + AsyncSession configuration. Start with the free tier to try SQLModel ORM code generation.