MUmerRazzaq avatar

sqlmodel-production

Production-ready SQLModel patterns for FastAPI applications. This skill should be used whencreating

作者 MUmerRazzaq|オープンソース

SQLModel Production Patterns

What This Skill Does

  • Creates production-grade SQLModel models with proper patterns
  • Implements relationships (one-to-many, many-to-many)
  • Configures session management (sync and async)
  • Builds FastAPI CRUD endpoints with SQLModel
  • Troubleshoots common database problems

What This Skill Does NOT Do

  • Set up Alembic migrations (separate concern)
  • Configure specific database servers (PostgreSQL, MySQL setup)
  • Handle authentication/authorization logic

Before Implementation

SourceGather
CodebaseExisting models, engine setup, project structure
ConversationUser's specific entities, relationships, constraints
Skill ReferencesPatterns from references/ for domain expertise
User GuidelinesProject naming conventions, async/sync preference

Quick Reference: Model Patterns

The Four Model Types

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


Quick Reference: Relationships

One-to-Many

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")

Many-to-Many (Link Table)

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.


Quick Reference: Session Management

Sync (Standard)

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

Async

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.


Quick Reference: CRUD Operations

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.


Database Problems Quick Reference

ProblemSymptomSolution
N+1 QueriesSlow list endpointsselectinload() / joinedload()
Connection Leak"Too many connections"Use context managers, pool_pre_ping=True
Detached Instance"not bound to Session"session.refresh() after commit
Transaction DeadlockTimeout errorsConsistent ordering, short transactions
Stale DataWrong values returnedsession.expire_all() or new session

See references/database-problems.md for complete troubleshooting guide.


Field Configuration

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.


Critical Rules

  1. Always separate table and non-table models - Base, Create, Read, Update pattern
  2. Never use echo=True in production - Logs all SQL, performance impact
  3. Always use pool_pre_ping=True - Validates connections before use
  4. Always index foreign keys - Critical for join performance
  5. Use back_populates not backref - Explicit is better than implicit
  6. Call session.refresh() after commit - Gets database-generated values
  7. Use model_validate() for creation - Proper Pydantic validation
  8. Use model_dump(exclude_unset=True) for updates - Only update provided fields
  9. Never commit inside loops - Batch operations for performance
  10. Always handle HTTPException for not found - Return 404, not 500