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. Use when this capability is needed.
metadata	{"author":"cyr-ius"}

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 ():
    

    name:  |  = 
    secret_name:  |  = 
    age:  |  = 

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(, db_hero., db_hero.name)
     db_hero





 () -> [HeroPublic]:
    
    heroes = session.(select(Hero).offset(offset).limit(limit)).()
     (heroes)





 () -> HeroPublic:
    
    hero = session.get(Hero, hero_id)
      hero:
         HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=,
        )
     hero





 () -> HeroPublic:
    
    db_hero = session.get(Hero, hero_id)
      db_hero:
         HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=,
        )

    
    hero_data = hero.model_dump(exclude_unset=)
    db_hero.sqlmodel_update(hero_data)

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

    logger.info(, db_hero.)
     db_hero





 () -> :
    
    hero = session.get(Hero, hero_id)
      hero:
         HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=,
        )

    session.delete(hero)
    session.commit()
    logger.info(, 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=)


 (HeroBase, table=):
    :  |  = Field(default=, primary_key=)

    
    team: Team |  = Relationship(back_populates=)

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  func
    result = session.(select(func.count()).select_from(Hero))
     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(
        ,
        json={: , : },
    )
     response.status_code == 
    data = response.json()
     data[] == 
     data[]   


 () -> :
    
    response = client.get()
     response.status_code == 


 () -> :
    
    create_resp = client.post(
        ,
        json={: , : },
    )
    hero_id = create_resp.json()[]

    update_resp = client.patch(, json={: })
     update_resp.status_code == 
     update_resp.json()[] == 
     update_resp.json()[] ==   

---

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/

---

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