Sqlmodel

SQLModel Skill

SQLModel is a Python library for interacting with SQL databases using Python objects. It combines Pydantic v2 (data validation) and SQLAlchemy (ORM) into a unified, type-safe API. It is created by the same author as FastAPI and designed to work seamlessly with it.

When to Use This Skill

• Adding a SQL database (SQLite, PostgreSQL, MySQL) to a FastAPI project
• Defining ORM models that also serve as Pydantic schemas
• Creating CRUD endpoints with proper request/response validation
• Managing database sessions and migrations
• Implementing relationships (one-to-many, many-to-many) between entities
• Replacing raw JSON file storage with a proper database layer

Installation

# SQLite (built-in, no extra driver needed)
pip install sqlmodel

# PostgreSQL (async)
pip install sqlmodel asyncpg

# PostgreSQL (sync)
pip install sqlmodel psycopg2-binary

Core Concepts

1. Model Types

SQLModel has two kinds of models:

Use table models for database entities and data models for API request/response schemas.

2. Inheritance Pattern (recommended)

Avoid field duplication by using a base data model:

HeroBase (data model — shared fields)
    ├── Hero (table=True — adds id as primary key)
    ├── HeroCreate (data model — for POST requests, no id)
    ├── HeroPublic (data model — for GET responses, id required)
    └── HeroUpdate (data model — all fields optional, for PATCH)

Complete FastAPI + SQLModel Example

Project Structure

app/
├── core/
│   └── database.py          # Engine and session dependency
├── models/
│   └── hero.py              # SQLModel models (table + data)
├── routers/
│   └── heroes.py            # FastAPI router with CRUD endpoints
└── main.py                  # FastAPI app with lifespan

Pattern 1: Database Engine and Session

# app/core/database.py
"""Database engine configuration and session dependency."""

from sqlmodel import SQLModel, Session, create_engine
from ..config import get_settings

settings = get_settings()

# SQLite (development)
connect_args = {"check_same_thread": False}  # required for SQLite only
engine = create_engine(
    settings.database_url,
    echo=False,  # set True to log SQL queries during development
    connect_args=connect_args,
)

# PostgreSQL (production) — no connect_args needed
# engine = create_engine(settings.database_url, echo=False)


def create_db_and_tables() -> None:
    """Create all tables defined by SQLModel models with table=True."""
    SQLModel.metadata.create_all(engine)


def get_session():
    """FastAPI dependency: yields a database session for the current request."""
    with Session(engine) as session:
        yield session

Pattern 2: Models with Inheritance

# app/models/hero.py
"""Hero SQLModel models — table model + API data models."""

from sqlmodel import Field, SQLModel


# ── Base model (shared fields, data model only) ───────────────────────────────

class HeroBase(SQLModel):
    """Shared fields inherited by all Hero variants."""

    name: str = Field(index=True, min_length=1, max_length=100)
    secret_name: str
    age: int | None = Field(default=None, ge=0, le=150)


# ── Table model (maps to the 'hero' table in the database) ───────────────────

class Hero(HeroBase, table=True):
    """Hero table model — stored in the database."""

    id: int | None = Field(default=None, primary_key=True)


# ── Data models (API schemas only, not stored in the database) ───────────────

class HeroCreate(HeroBase):
    """Request schema for creating a new hero (no id — generated by the DB)."""

    pass


class HeroPublic(HeroBase):
    """Response schema for returning a hero (id is always present)."""

    id: int  # required (not optional) because the DB always assigns an id


class HeroUpdate(SQLModel):
    """Request schema for partial update (PATCH) — all fields are optional."""

    name: str | None = None
    secret_name: str | None = None
    age: int | None = None

Pattern 3: CRUD Router

# app/routers/heroes.py
"""Hero CRUD endpoints using SQLModel and FastAPI."""

import logging
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlmodel import Session, select

from ..core.database import get_session
from ..models.hero import Hero, HeroCreate, HeroPublic, HeroUpdate

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/heroes", tags=["Heroes"])

# Shortcut type alias for the session dependency
SessionDep = Annotated[Session, Depends(get_session)]


# ── Create ────────────────────────────────────────────────────────────────────

@router.post("", response_model=HeroPublic, status_code=status.HTTP_201_CREATED)
def create_hero(
    *,
    session: SessionDep,
    hero: HeroCreate,
) -> HeroPublic:
    """Create a new hero and persist it to the database.

    Args:
        session: Database session (injected by FastAPI).
        hero: Validated hero creation data.

    Returns:
        HeroPublic: The created hero with its database-assigned id.

    Raises:
        HTTPException 409: If a hero with the same name already exists.
    """
    # Check for duplicate name
    existing = session.exec(select(Hero).where(Hero.name == hero.name)).first()
    if existing:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"A hero named '{hero.name}' already exists",
        )

    # Convert the data model (HeroCreate) to the table model (Hero)
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)  # reload to get the auto-generated id

    logger.info("Hero created: id=%s name=%s", db_hero.id, db_hero.name)
    return db_hero


# ── Read (list) ───────────────────────────────────────────────────────────────

@router.get("", response_model=list[HeroPublic])
def read_heroes(
    *,
    session: SessionDep,
    offset: int = 0,
    limit: int = Query(default=20, le=100),
) -> list[HeroPublic]:
    """Return a paginated list of heroes.

    Args:
        session: Database session.
        offset: Number of rows to skip (for pagination).
        limit: Maximum rows to return (capped at 100).

    Returns:
        list[HeroPublic]: Heroes matching the pagination parameters.
    """
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return list(heroes)


# ── Read (one) ────────────────────────────────────────────────────────────────

@router.get("/{hero_id}", response_model=HeroPublic)
def read_hero(
    *,
    session: SessionDep,
    hero_id: int,
) -> HeroPublic:
    """Return a single hero by its id.

    Args:
        session: Database session.
        hero_id: Primary key of the hero to retrieve.

    Returns:
        HeroPublic: The requested hero.

    Raises:
        HTTPException 404: If no hero with the given id exists.
    """
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Hero with id={hero_id} not found",
        )
    return hero


# ── Update (partial) ──────────────────────────────────────────────────────────

@router.patch("/{hero_id}", response_model=HeroPublic)
def update_hero(
    *,
    session: SessionDep,
    hero_id: int,
    hero: HeroUpdate,
) -> HeroPublic:
    """Partially update a hero (only fields provided in the request are changed).

    Args:
        session: Database session.
        hero_id: Primary key of the hero to update.
        hero: Fields to update (all optional).

    Returns:
        HeroPublic: The updated hero.

    Raises:
        HTTPException 404: If no hero with the given id exists.
    """
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Hero with id={hero_id} not found",
        )

    # Only update the fields that were explicitly sent in the request
    hero_data = hero.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)

    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)

    logger.info("Hero updated: id=%s", db_hero.id)
    return db_hero


# ── Delete ────────────────────────────────────────────────────────────────────

@router.delete("/{hero_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_hero(
    *,
    session: SessionDep,
    hero_id: int,
) -> None:
    """Delete a hero by its id.

    Args:
        session: Database session.
        hero_id: Primary key of the hero to delete.

    Raises:
        HTTPException 404: If no hero with the given id exists.
    """
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Hero with id={hero_id} not found",
        )

    session.delete(hero)
    session.commit()
    logger.info("Hero deleted: id=%s", hero_id)

Pattern 4: FastAPI App with Lifespan

# app/main.py
"""FastAPI application entry point with SQLModel database initialization."""

from contextlib import asynccontextmanager

from fastapi import FastAPI

from .core.database import create_db_and_tables
from .routers import heroes


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Create database tables on startup."""
    create_db_and_tables()
    yield


app = FastAPI(title="Hero API", lifespan=lifespan)

app.include_router(heroes.router, prefix="/api")

Relationships

One-to-Many (Team → Heroes)

# app/models/team.py
"""Team and Hero models with a one-to-many relationship."""

from typing import TYPE_CHECKING
from sqlmodel import Field, Relationship, SQLModel

# Avoid circular imports at runtime — only used for type checking
if TYPE_CHECKING:
    from .hero import Hero


class TeamBase(SQLModel):
    """Shared fields for all Team variants."""

    name: str = Field(index=True)
    headquarters: str


class Team(TeamBase, table=True):
    """Team table model."""

    id: int | None = Field(default=None, primary_key=True)

    # Relationship: one team has many heroes
    # back_populates must match the attribute name on the other side
    heroes: list["Hero"] = Relationship(back_populates="team")


# app/models/hero.py — updated to include the relationship
class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = None
    # Foreign key to the team table
    team_id: int | None = Field(default=None, foreign_key="team.id")


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

    # Relationship: each hero belongs to one team (or none)
    team: Team | None = Relationship(back_populates="heroes")

Many-to-Many (Heroes ↔ Powers via link table)

# app/models/power.py
"""Many-to-many relationship between Hero and Power via HeroPowerLink."""

from sqlmodel import Field, Relationship, SQLModel


class HeroPowerLink(SQLModel, table=True):
    """Link table for the Hero <-> Power many-to-many relationship."""

    hero_id: int | None = Field(
        default=None, foreign_key="hero.id", primary_key=True
    )
    power_id: int | None = Field(
        default=None, foreign_key="power.id", primary_key=True
    )


class PowerBase(SQLModel):
    """Shared fields for all Power variants."""

    name: str = Field(index=True)
    description: str | None = None


class Power(PowerBase, table=True):
    """Power table model."""

    id: int | None = Field(default=None, primary_key=True)

    # Many-to-many: a power can belong to many heroes
    heroes: list["Hero"] = Relationship(
        back_populates="powers", link_model=HeroPowerLink
    )

Advanced Queries

Filtering, Ordering, and Pagination

from sqlmodel import Session, select, and_, or_, col


def search_heroes(
    session: Session,
    name_filter: str | None = None,
    min_age: int | None = None,
    max_age: int | None = None,
    offset: int = 0,
    limit: int = 20,
) -> list[Hero]:
    """Search heroes with optional filters, ordering, and pagination."""

    statement = select(Hero)

    # Build conditions dynamically
    conditions = []
    if name_filter:
        conditions.append(col(Hero.name).contains(name_filter))
    if min_age is not None:
        conditions.append(Hero.age >= min_age)
    if max_age is not None:
        conditions.append(Hero.age <= max_age)

    if conditions:
        statement = statement.where(and_(*conditions))

    # Order by name ascending, then by id
    statement = statement.order_by(Hero.name, Hero.id)

    # Apply pagination
    statement = statement.offset(offset).limit(limit)

    return list(session.exec(statement).all())


def count_heroes(session: Session) -> int:
    """Return the total number of heroes in the database."""
    from sqlmodel import func
    result = session.exec(select(func.count()).select_from(Hero))
    return result.one()

Querying with Relationships (JOIN)

def get_heroes_for_team(session: Session, team_id: int) -> list[Hero]:
    """Return all heroes belonging to a specific team."""
    statement = select(Hero).where(Hero.team_id == team_id)
    return list(session.exec(statement).all())


def get_team_with_heroes(session: Session, team_id: int) -> Team | None:
    """Return a team and eagerly load its heroes."""
    from sqlmodel import selectinload

    statement = (
        select(Team)
        .where(Team.id == team_id)
        .options(selectinload(Team.heroes))  # eager load to avoid N+1
    )
    return session.exec(statement).first()

Database Configuration (pydantic-settings)

# app/config.py
"""Application settings loaded from environment variables."""

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    """Database and application settings."""

    # SQLite (development): sqlite:///./database.db
    # PostgreSQL (production): postgresql+psycopg2://user:pass@host:5432/dbname
    database_url: str = "sqlite:///./database.db"

    # Controls SQLAlchemy query logging (disable in production)
    database_echo: bool = False

    class Config:
        env_file = ".env"
        case_sensitive = False

Environment variables:

# .env — development (SQLite)
DATABASE_URL=sqlite:///./database.db
DATABASE_ECHO=false

# .env.production — production (PostgreSQL)
DATABASE_URL=postgresql+psycopg2://portalcrane:secret@db:5432/portalcrane
DATABASE_ECHO=false

Testing with SQLModel

# tests/conftest.py
"""Pytest fixtures for SQLModel + FastAPI integration tests."""

import pytest
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, Session, StaticPool, create_engine

from app.core.database import get_session
from app.main import app


@pytest.fixture(name="session")
def session_fixture():
    """Create an in-memory SQLite engine for each test."""
    engine = create_engine(
        "sqlite://",  # in-memory SQLite
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,  # share the same connection across threads
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    """Override the get_session dependency to use the test database."""

    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


# tests/test_heroes.py
def test_create_hero(client: TestClient) -> None:
    """Creating a hero returns 201 and the hero with an id."""
    response = client.post(
        "/api/heroes",
        json={"name": "Deadpond", "secret_name": "Dive Wilson"},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Deadpond"
    assert data["id"] is not None


def test_read_hero_not_found(client: TestClient) -> None:
    """Reading a non-existent hero returns 404."""
    response = client.get("/api/heroes/9999")
    assert response.status_code == 404


def test_update_hero(client: TestClient) -> None:
    """Partial update only changes the specified fields."""
    create_resp = client.post(
        "/api/heroes",
        json={"name": "Spider-Boy", "secret_name": "Pedro"},
    )
    hero_id = create_resp.json()["id"]

    update_resp = client.patch(f"/api/heroes/{hero_id}", json={"age": 25})
    assert update_resp.status_code == 200
    assert update_resp.json()["age"] == 25
    assert update_resp.json()["name"] == "Spider-Boy"  # unchanged

Common Patterns and Best Practices

1. Always Use model_dump(exclude_unset=True) for PATCH

# ✅ CORRECT — only updates the fields the client actually sent
hero_data = hero_update.model_dump(exclude_unset=True)
db_hero.sqlmodel_update(hero_data)

# ❌ WRONG — would reset optional fields to None
hero_data = hero_update.model_dump()

2. Always refresh() After commit() to Get Generated Values

session.add(db_hero)
session.commit()
session.refresh(db_hero)  # ← required to read the auto-generated id and defaults
return db_hero

3. Use session.get() for Primary Key Lookups

# ✅ CORRECT — efficient primary key lookup
hero = session.get(Hero, hero_id)

# ❌ AVOID — unnecessary SQL query builder for simple PK lookups
hero = session.exec(select(Hero).where(Hero.id == hero_id)).first()

4. Use model_validate() to Convert Between Model Types

# Convert HeroCreate (data model) to Hero (table model)
db_hero = Hero.model_validate(hero_create)

# Convert an ORM instance to a Pydantic response schema
hero_public = HeroPublic.model_validate(db_hero)

5. Never Inherit from Table Models

# ✅ CORRECT — inherit from a data model base
class HeroBase(SQLModel): ...          # base data model
class Hero(HeroBase, table=True): ...  # table model inherits from data model
class HeroCreate(HeroBase): ...        # data model inherits from data model

# ❌ WRONG — never inherit from a table model
class HeroAdmin(Hero, table=True): ... # creates a new table unintentionally

6. Handle Session Errors with try/except

from sqlalchemy.exc import IntegrityError


def create_hero_safe(session: Session, hero: HeroCreate) -> Hero:
    """Create a hero with database-level duplicate detection."""
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Hero already exists (database constraint violation)",
        )
    session.refresh(db_hero)
    return db_hero

Quick Reference

Resources

• Official documentation: https://sqlmodel.tiangolo.com
• Source code: https://github.com/fastapi/sqlmodel
• FastAPI integration tutorial: https://sqlmodel.tiangolo.com/tutorial/fastapi/
• Relationships: https://sqlmodel.tiangolo.com/tutorial/relationship-attributes/
• Many-to-many: https://sqlmodel.tiangolo.com/tutorial/many-to-many/
• Testing: https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/
• Advanced (UUID, Decimal): https://sqlmodel.tiangolo.com/advanced/

Source: cyr-ius/wireguard-ui — distributed by TomeVault.