sqlmodel

name	sqlmodel
description	Build SQL database integrations with SQLModel for FastAPI projects. Use when working with databases in Python, defining ORM models, creating CRUD operations, managing sessions, or integrating SQL databases with FastAPI. SQLModel combines Pydantic v2 and SQLAlchemy into a single unified API.

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:

Type	table=True	Stored in DB	Pydantic model	SQLAlchemy model
Table model	✅ Yes	✅ Yes	✅ Yes	✅ Yes
Data model	❌ No	❌ No	✅ Yes	❌ No

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

Operation	SQLModel Code
Create table model	class Item(SQLModel, table=True): ...
Create data model	class ItemCreate(SQLModel): ...
Primary key	id: int | None = Field(default=None, primary_key=True)
Foreign key	team_id: int | None = Field(default=None, foreign_key="team.id")
Indexed field	name: str = Field(index=True)
Unique field	email: str = Field(unique=True)
Session dependency	def get_session(): yield Session(engine)
Insert	session.add(obj); session.commit(); session.refresh(obj)
Select all	session.exec(select(Model)).all()
Select by PK	session.get(Model, id)
Filter	select(Model).where(Model.field == value)
Partial update	obj.sqlmodel_update(data.model_dump(exclude_unset=True))
Delete	session.delete(obj); session.commit()
Count	session.exec(select(func.count()).select_from(Model)).one()

---

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/