Production-ready SQLModel patterns for FastAPI applications...
| Source | Gather |
|---|---|
| Codebase | Existing models, engine setup, project structure |
| Conversation | User's specific entities, relationships, constraints |
| Skill References | Patterns from references/ for domain expertise |
| User Guidelines | Project naming conventions, async/sync preference |
# 1. Base - Shared fields (NO table=True)
class HeroBase(SQLModel):
name: str = Field(max_length=100, index=True)
age: int | None = Field(default=None, index=True)
# 2. Table - Database table (table=True)
class Hero(HeroBase, table=True):
id: int | None = Field(default=None, primary_key=True)
secret_name: str = Field(max_length=255)
# 3. Create - Input validation (NO id)
class HeroCreate(HeroBase):
secret_name: str
# 4. Read/Public - API response (includes id)
class HeroPublic(HeroBase):
id: int
# 5. Update - Partial updates (all Optional)
class HeroUpdate(SQLModel):
name: str | None = None
age: int | None = None
See references/model-patterns.md for complete patterns including audit fields.
class Team(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
heroes: list["Hero"] = Relationship(back_populates="team")
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
team_id: int | None = Field(default=None, foreign_key="team.id", index=True)
team: Team | None = Relationship(back_populates="heroes")
class HeroTeamLink(SQLModel, table=True):
team_id: int = Field(foreign_key="team.id", primary_key=True)
hero_id: int = Field(foreign_key="hero.id", primary_key=True)
role: str = Field(default="member") # Extra data on link
class Team(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
heroes: list["Hero"] = Relationship(back_populates="teams", link_model=HeroTeamLink)
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
teams: list[Team] = Relationship(back_populates="heroes", link_model=HeroTeamLink)
See references/relationships.md for advanced patterns.
from sqlmodel import Session, create_engine
engine = create_engine(DATABASE_URL, echo=False, pool_pre_ping=True)
def get_session():
with Session(engine) as session:
yield session
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
async_engine = create_async_engine(
DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://"),
echo=False,
pool_pre_ping=True,
)
async def get_async_session():
async with AsyncSession(async_engine) as session:
yield session
See references/session-management.md for connection pooling configuration.
from fastapi import Depends, HTTPException, Query
from sqlmodel import select
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
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(
*,
session: Session = Depends(get_session),
offset: int = 0,
limit: int = Query(default=100, le=100),
):
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
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(*, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate):
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(*, session: Session = Depends(get_session), hero_id: int):
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}
See references/crud-operations.md for async patterns and bulk operations.
| Problem | Symptom | Solution |
|---|---|---|
| N+1 Queries | Slow list endpoints | selectinload() / joinedload() |
| Connection Leak | "Too many connections" | Use context managers, pool_pre_ping=True |
| Detached Instance | "not bound to Session" | session.refresh() after commit |
| Transaction Deadlock | Timeout errors | Consistent ordering, short transactions |
| Stale Data | Wrong values returned | session.expire_all() or new session |
See references/database-problems.md for complete troubleshooting guide.
from sqlmodel import Field
class Hero(SQLModel, table=True):
# Primary key
id: int | None = Field(default=None, primary_key=True)
# Indexed for faster queries
name: str = Field(index=True, max_length=100)
# Unique constraint
email: str = Field(unique=True, max_length=255)
# Foreign key with index
team_id: int | None = Field(default=None, foreign_key="team.id", index=True)
# Nullable with default
age: int | None = Field(default=None)
# Required field with validation
secret_name: str = Field(min_length=1, max_length=255)
See references/field-configuration.md for advanced constraints and custom types.
echo=True in production - Logs all SQL, performance impactpool_pre_ping=True - Validates connections before useback_populates not backref - Explicit is better than implicitsession.refresh() after commit - Gets database-generated valuesmodel_validate() for creation - Proper Pydantic validationmodel_dump(exclude_unset=True) for updates - Only update provided fieldsHTTPException for not found - Return 404, not 500