Все важные темы SQLModel — полная карта

Общая картина

SQLModel
├── 1.  Определение моделей (Models)
├── 2.  Типы полей и валидация (Fields & Validation)
├── 3.  Связи (Relationships) ✅ уже разобрали
├── 4.  Engine и подключение к БД
├── 5.  Session и Unit of Work
├── 6.  CRUD-операции
├── 7.  Запросы (select, where, join, group_by...)
├── 8.  Миграции (Alembic)
├── 9.  Паттерн "множественных моделей" (Create/Read/Update)
├── 10. Индексы и ограничения (Constraints)
├── 11. Асинхронность (async)
├── 12. Интеграция с FastAPI
├── 13. События и хуки (Events)
├── 14. Наследование моделей
├── 15. Работа с JSON/ARRAY полями
├── 16. Raw SQL и гибридные свойства
├── 17. Тестирование
└── 18. Производительность и оптимизация

1. Определение моделей

from sqlmodel import SQLModel, Field
from datetime import datetime, date
from decimal import Decimal
from enum import Enum
import uuid

# --- Базовая модель (не таблица) ---
class UserBase(SQLModel):
    """Pydantic-модель для валидации, без таблицы в БД"""
    name: str
    email: str

# --- Табличная модель ---
class User(UserBase, table=True):
    """Реальная таблица в БД"""
    __tablename__ = "users"  # кастомное имя таблицы

    id: int | None = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)

# --- Модель с UUID ---
class Item(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    name: str

# --- Модель с Enum ---
class Status(str, Enum):
    active = "active"
    inactive = "inactive"
    banned = "banned"

class Account(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    status: Status = Field(default=Status.active)

2. Типы полей и валидация

from sqlmodel import SQLModel, Field
from pydantic import field_validator, model_validator
from sqlalchemy import Column, String, Text, Numeric

class Product(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)

    # --- Строки ---
    name: str = Field(min_length=1, max_length=100, index=True)
    slug: str = Field(max_length=100, unique=True)
    description: str = Field(
        sa_column=Column(Text)  # TEXT вместо VARCHAR
    )

    # --- Числа ---
    price: Decimal = Field(
        max_digits=10,
        decimal_places=2,
        ge=0,  # >= 0
    )
    quantity: int = Field(default=0, ge=0, le=999999)

    # --- Булевы ---
    is_active: bool = Field(default=True)

    # --- Nullable ---
    sku: str | None = Field(default=None, max_length=50)

    # --- Со значением по умолчанию ---
    rating: float = Field(default=0.0, ge=0, le=5)

    # --- Кастомная колонка через SQLAlchemy ---
    code: str = Field(
        sa_column=Column(String(10), nullable=False, server_default="NEW")
    )

    # --- Pydantic-валидаторы работают! ---
    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be blank")
        return v.strip()

    @field_validator("slug")
    @classmethod
    def slug_format(cls, v: str) -> str:
        import re
        if not re.match(r'^[a-z0-9-]+$', v):
            raise ValueError("Slug must be lowercase alphanumeric with hyphens")
        return v

    @model_validator(mode="after")
    def check_consistency(self):
        if self.price == 0 and self.quantity > 0:
            raise ValueError("Free products cannot have stock")
        return self

Маппинг типов Python → SQL

4. Engine и подключение

from sqlmodel import create_engine, SQLModel

# --- SQLite ---
engine = create_engine(
    "sqlite:///database.db",
    echo=True,           # логировать SQL
    connect_args={"check_same_thread": False},  # для FastAPI
)

# --- PostgreSQL ---
engine = create_engine(
    "postgresql://user:password@localhost:5432/dbname",
    echo=False,
    pool_size=20,         # размер пула
    max_overflow=10,      # доп. соединения сверх пула
    pool_timeout=30,      # ожидание свободного соединения
    pool_recycle=1800,    # пересоздание соединения каждые 30 мин
    pool_pre_ping=True,   # проверка соединения перед использованием
)

# --- PostgreSQL через asyncpg ---
# pip install asyncpg
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/dbname"
)

# --- MySQL ---
engine = create_engine(
    "mysql+pymysql://user:password@localhost:3306/dbname"
)

# --- Из переменных окружения ---
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///db.sqlite3"
    db_echo: bool = False

    class Config:
        env_file = ".env"

settings = Settings()
engine = create_engine(settings.database_url, echo=settings.db_echo)

# --- Создание таблиц ---
SQLModel.metadata.create_all(engine)

5. Session и Unit of Work

from sqlmodel import Session, select

# --- Базовое использование ---
with Session(engine) as session:
    hero = Hero(name="Batman")
    session.add(hero)
    session.commit()
    session.refresh(hero)  # обновить объект из БД (получить id)
    print(hero.id)

# --- Транзакции ---
with Session(engine) as session:
    try:
        session.add(Hero(name="A"))
        session.add(Hero(name="B"))
        session.commit()  # обе записи или ни одной
    except Exception:
        session.rollback()
        raise

# --- Вложенные транзакции (savepoints) ---
with Session(engine) as session:
    session.add(Hero(name="Safe"))
    session.flush()  # отправить в БД без коммита

    session.begin_nested()  # savepoint
    try:
        session.add(Hero(name="Risky"))
        session.flush()
        raise ValueError("oops")
    except ValueError:
        session.rollback()  # откат до savepoint, "Safe" останется

    session.commit()

# --- Основные методы Session ---
session.add(obj)           # Добавить объект
session.add_all([a, b])    # Добавить несколько
session.delete(obj)        # Удалить
session.commit()           # Зафиксировать транзакцию
session.rollback()         # Откатить
session.refresh(obj)       # Обновить из БД
session.flush()            # Отправить в БД без коммита
session.get(Model, pk)     # Получить по PK (кэшируется)
session.exec(statement)    # Выполнить запрос
session.expire(obj)        # Пометить как устаревший
session.expunge(obj)       # Отсоединить от сессии
session.merge(obj)         # Слить объект в сессию

6. CRUD-операции

from sqlmodel import Session, select

# ═══════════════ CREATE ═══════════════
with Session(engine) as session:
    hero = Hero(name="Spider-Man", age=25)
    session.add(hero)
    session.commit()
    session.refresh(hero)

    # Массовое создание
    heroes = [Hero(name=f"Hero-{i}") for i in range(100)]
    session.add_all(heroes)
    session.commit()

# ═══════════════ READ ═══════════════
with Session(engine) as session:
    # По ID
    hero = session.get(Hero, 1)

    # Один результат
    stmt = select(Hero).where(Hero.name == "Spider-Man")
    hero = session.exec(stmt).first()       # None если нет
    hero = session.exec(stmt).one()          # Ошибка если нет или > 1
    hero = session.exec(stmt).one_or_none()  # None или ошибка если > 1

    # Все результаты
    heroes = session.exec(select(Hero)).all()

# ═══════════════ UPDATE ═══════════════
with Session(engine) as session:
    hero = session.get(Hero, 1)
    hero.name = "New Name"
    hero.age = 30
    session.add(hero)
    session.commit()
    session.refresh(hero)

    # Обновление из словаря (Pydantic)
    update_data = HeroUpdate(name="Updated")
    hero_data = update_data.model_dump(exclude_unset=True)
    for key, value in hero_data.items():
        setattr(hero, key, value)
    session.add(hero)
    session.commit()

# ═══════════════ DELETE ═══════════════
with Session(engine) as session:
    hero = session.get(Hero, 1)
    session.delete(hero)
    session.commit()

7. Запросы (самая обширная тема)

from sqlmodel import select, or_, and_, not_, col, func, text

# ═══════════════ WHERE ═══════════════
select(Hero).where(Hero.name == "Batman")
select(Hero).where(Hero.age >= 18)
select(Hero).where(Hero.age != None)  # noqa
select(Hero).where(Hero.name.contains("man"))
select(Hero).where(Hero.name.startswith("B"))
select(Hero).where(Hero.name.endswith("man"))
select(Hero).where(Hero.name.in_(["Batman", "Superman"]))
select(Hero).where(Hero.name.not_in(["Joker"]))
select(Hero).where(Hero.age.between(18, 30))
select(Hero).where(Hero.name.like("%man%"))
select(Hero).where(Hero.name.ilike("%man%"))  # case-insensitive

# --- Комбинированные условия ---
select(Hero).where(Hero.age >= 18, Hero.age <= 30)         # AND (неявный)
select(Hero).where(and_(Hero.age >= 18, Hero.age <= 30))   # AND (явный)
select(Hero).where(or_(Hero.name == "A", Hero.name == "B"))
select(Hero).where(not_(Hero.age < 18))

# ═══════════════ ORDER BY ═══════════════
select(Hero).order_by(Hero.name)                    # ASC
select(Hero).order_by(Hero.name.desc())             # DESC
select(Hero).order_by(Hero.age.desc(), Hero.name)   # несколько

# ═══════════════ LIMIT / OFFSET ═══════════════
select(Hero).offset(10).limit(20)  # пагинация

# ═══════════════ DISTINCT ═══════════════
select(Hero.name).distinct()

# ═══════════════ GROUP BY + HAVING ═══════════════
from sqlmodel import func

statement = (
    select(Hero.team_id, func.count(Hero.id).label("cnt"))
    .group_by(Hero.team_id)
    .having(func.count(Hero.id) > 3)
)

# ═══════════════ AGGREGATE ═══════════════
session.exec(select(func.count()).select_from(Hero)).one()
session.exec(select(func.max(Hero.age))).one()
session.exec(select(func.min(Hero.age))).one()
session.exec(select(func.avg(Hero.age))).one()
session.exec(select(func.sum(Hero.age))).one()

# ═══════════════ JOIN ═══════════════
# Автоматический (по FK)
select(Hero, Team).join(Team)
select(Hero, Team).join(Team, isouter=True)  # LEFT JOIN

# Явный
select(Hero, Team).join(Team, Hero.team_id == Team.id)

# Выбор конкретных полей
select(Hero.name, Team.name).join(Team)

# ═══════════════ SUBQUERY ═══════════════
subq = select(func.avg(Hero.age)).scalar_subquery()
statement = select(Hero).where(Hero.age > subq)

# ═══════════════ RAW SQL ═══════════════
from sqlmodel import text

with Session(engine) as session:
    result = session.exec(
        text("SELECT * FROM hero WHERE age > :age"),
        params={"age": 18}
    )
    for row in result:
        print(row)

# ═══════════════ EXISTS ═══════════════
from sqlalchemy import exists

subq = select(Hero).where(Hero.team_id == Team.id).exists()
statement = select(Team).where(subq)

# ═══════════════ CASE ═══════════════
from sqlalchemy import case

statement = select(
    Hero.name,
    case(
        (Hero.age < 18, "minor"),
        (Hero.age < 65, "adult"),
        else_="senior"
    ).label("category")
)

8. Миграции (Alembic)

# Установка
pip install alembic

# Инициализация
alembic init alembic

alembic/env.py — ключевые изменения:

from sqlmodel import SQLModel
from app.models import *  # импортировать ВСЕ модели!

target_metadata = SQLModel.metadata  # ← вместо None

# Создать миграцию
alembic revision --autogenerate -m "create users table"

# Применить
alembic upgrade head

# Откатить
alembic downgrade -1

# Посмотреть историю
alembic history

# Текущая версия
alembic current

Пример сгенерированной миграции:

def upgrade():
    op.create_table(
        'hero',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(), nullable=False),
        sa.Column('age', sa.Integer(), nullable=True),
        sa.PrimaryKeyConstraint('id'),
    )
    op.create_index('ix_hero_name', 'hero', ['name'])

def downgrade():
    op.drop_index('ix_hero_name', 'hero')
    op.drop_table('hero')

9. Паттерн множественных моделей

# ═══════ Base (общие поля) ═══════
class HeroBase(SQLModel):
    name: str = Field(min_length=1, max_length=100)
    age: int | None = Field(default=None, ge=0)
    team_id: int | None = None

# ═══════ Table (БД) ═══════
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    team: "Team | None" = Relationship(back_populates="heroes")

# ═══════ Create (входные данные) ═══════
class HeroCreate(HeroBase):
    pass  # наследует name, age, team_id

# ═══════ Read (ответ API) ═══════
class HeroRead(HeroBase):
    id: int
    created_at: datetime

# ═══════ Read with relations ═══════
class HeroReadFull(HeroRead):
    team: "TeamRead | None" = None

# ═══════ Update (частичное обновление) ═══════
class HeroUpdate(SQLModel):
    name: str | None = None      # все поля Optional
    age: int | None = None
    team_id: int | None = None

# ═══════ List response ═══════
class HeroListResponse(SQLModel):
    data: list[HeroRead]
    total: int
    page: int
    per_page: int

10. Индексы и ограничения

from sqlmodel import SQLModel, Field
from sqlalchemy import UniqueConstraint, Index, CheckConstraint

class Product(SQLModel, table=True):
    __table_args__ = (
        # Составной уникальный ключ
        UniqueConstraint("sku", "warehouse_id", name="uq_sku_warehouse"),

        # Составной индекс
        Index("ix_category_name", "category", "name"),

        # Check constraint
        CheckConstraint("price >= 0", name="ck_positive_price"),
        CheckConstraint("quantity >= 0", name="ck_positive_qty"),
    )

    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)               # простой индекс
    sku: str = Field(max_length=50)
    warehouse_id: int
    category: str
    price: float = Field(ge=0)
    quantity: int = Field(ge=0)
    email: str = Field(unique=True)              # уникальное поле

11. Асинхронность (async)

# pip install sqlmodel aiosqlite  (или asyncpg для PostgreSQL)

from sqlmodel import SQLModel, Field, select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker

# --- Engine ---
async_engine = create_async_engine(
    "sqlite+aiosqlite:///database.db",
    # "postgresql+asyncpg://user:pass@localhost/db",
    echo=True,
)

# --- Создание таблиц ---
async def create_db():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# --- Session factory ---
async_session_factory = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# --- CRUD ---
async def create_hero(name: str) -> Hero:
    async with AsyncSession(async_engine) as session:
        hero = Hero(name=name)
        session.add(hero)
        await session.commit()
        await session.refresh(hero)
        return hero

async def get_heroes() -> list[Hero]:
    async with AsyncSession(async_engine) as session:
        result = await session.exec(select(Hero))
        return result.all()

async def get_hero(hero_id: int) -> Hero | None:
    async with AsyncSession(async_engine) as session:
        return await session.get(Hero, hero_id)

# --- FastAPI + async ---
from fastapi import FastAPI, Depends

app = FastAPI()

async def get_session():
    async with AsyncSession(async_engine) as session:
        yield session

@app.get("/heroes/")
async def read_heroes(session: AsyncSession = Depends(get_session)):
    heroes = await session.exec(select(Hero))
    return heroes.all()

# --- Eager loading (async) ---
from sqlalchemy.orm import selectinload

async def get_team_with_heroes(team_id: int):
    async with AsyncSession(async_engine) as session:
        statement = (
            select(Team)
            .where(Team.id == team_id)
            .options(selectinload(Team.heroes))
        )
        result = await session.exec(statement)
        return result.first()

12. Интеграция с FastAPI (продвинутая)

from fastapi import FastAPI, Depends, HTTPException, Query
from sqlmodel import Session, select, func
from contextlib import asynccontextmanager

# --- Lifespan ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    SQLModel.metadata.create_all(engine)
    yield
    # cleanup

app = FastAPI(lifespan=lifespan)

# --- Dependency ---
def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Depends(get_session)

# --- Пагинация ---
@app.get("/heroes/", response_model=list[HeroRead])
def list_heroes(
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=20, le=100),
    search: str | None = None,
    session: Session = SessionDep,
):
    statement = select(Hero)
    if search:
        statement = statement.where(Hero.name.ilike(f"%{search}%"))
    statement = statement.offset(offset).limit(limit)
    return session.exec(statement).all()

# --- Фильтрация + сортировка ---
@app.get("/heroes/advanced", response_model=list[HeroRead])
def list_heroes_advanced(
    min_age: int | None = None,
    max_age: int | None = None,
    team_id: int | None = None,
    sort_by: str = "name",
    sort_order: str = "asc",
    session: Session = SessionDep,
):
    statement = select(Hero)

    if min_age is not None:
        statement = statement.where(Hero.age >= min_age)
    if max_age is not None:
        statement = statement.where(Hero.age <= max_age)
    if team_id is not None:
        statement = statement.where(Hero.team_id == team_id)

    sort_column = getattr(Hero, sort_by, Hero.name)
    if sort_order == "desc":
        sort_column = sort_column.desc()
    statement = statement.order_by(sort_column)

    return session.exec(statement).all()

# --- PATCH (частичное обновление) ---
@app.patch("/heroes/{hero_id}", response_model=HeroRead)
def update_hero(
    hero_id: int,
    hero_update: HeroUpdate,
    session: Session = SessionDep,
):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")

    hero_data = hero_update.model_dump(exclude_unset=True)
    hero.sqlmodel_update(hero_data)  # SQLModel 0.0.14+
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero

13. События и хуки

from sqlalchemy import event
from sqlmodel import Session

# --- Before insert ---
@event.listens_for(Hero, "before_insert")
def hero_before_insert(mapper, connection, target):
    target.name = target.name.strip().title()

# --- After update ---
@event.listens_for(Hero, "after_update")
def hero_after_update(mapper, connection, target):
    print(f"Hero {target.id} was updated")

# --- Before delete ---
@event.listens_for(Hero, "before_delete")
def hero_before_delete(mapper, connection, target):
    print(f"About to delete hero {target.name}")

# --- Session events ---
@event.listens_for(Session, "after_commit")
def after_commit(session):
    print("Transaction committed")

# --- Валидация через set ---
@event.listens_for(Hero.age, "set")
def validate_age(target, value, oldvalue, initiator):
    if value is not None and value < 0:
        raise ValueError("Age cannot be negative")

14. Наследование моделей

Single Table Inheritance

from sqlmodel import SQLModel, Field

class Employee(SQLModel, table=True):
    __tablename__ = "employee"

    id: int | None = Field(default=None, primary_key=True)
    name: str
    type: str  # discriminator

    __mapper_args__ = {
        "polymorphic_on": "type",
        "polymorphic_identity": "employee",
    }

class Manager(Employee):
    department: str | None = None

    __mapper_args__ = {
        "polymorphic_identity": "manager",
    }

class Engineer(Employee):
    language: str | None = None

    __mapper_args__ = {
        "polymorphic_identity": "engineer",
    }

15. JSON и сложные типы

from sqlmodel import SQLModel, Field
from sqlalchemy import Column, JSON
from typing import Any

class Config(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

    # JSON поле
    settings: dict[str, Any] = Field(
        default_factory=dict,
        sa_column=Column(JSON),
    )

    tags: list[str] = Field(
        default_factory=list,
        sa_column=Column(JSON),
    )

# PostgreSQL ARRAY
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy import Column, String

class Post(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    tags: list[str] = Field(
        sa_column=Column(ARRAY(String))
    )

16. Raw SQL и гибридные свойства

from sqlalchemy.ext.hybrid import hybrid_property
from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    first_name: str
    last_name: str
    birth_year: int

    # Вычисляемое свойство (Python-уровень)
    @hybrid_property
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

    # SQL-уровень (можно использовать в WHERE)
    @full_name.expression
    @classmethod
    def full_name(cls):
        return cls.first_name + " " + cls.last_name

# Теперь можно:
select(User).where(User.full_name == "John Doe")

17. Тестирование

import pytest
from sqlmodel import SQLModel, Session, create_engine, select
from sqlmodel.pool import StaticPool

@pytest.fixture(name="session")
def session_fixture():
    """Новая in-memory БД для каждого теста"""
    engine = create_engine(
        "sqlite://",  # in-memory
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,  # ← важно для SQLite in-memory
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

def test_create_hero(session: Session):
    hero = Hero(name="Test Hero", age=25)
    session.add(hero)
    session.commit()
    session.refresh(hero)

    assert hero.id is not None
    assert hero.name == "Test Hero"

def test_read_heroes(session: Session):
    session.add_all([
        Hero(name="A"),
        Hero(name="B"),
    ])
    session.commit()

    heroes = session.exec(select(Hero)).all()
    assert len(heroes) == 2

# --- Тестирование FastAPI ---
from fastapi.testclient import TestClient

@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        yield session

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

def test_api_create_hero(client: TestClient):
    response = client.post("/heroes/", json={"name": "API Hero", "age": 30})
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "API Hero"
    assert "id" in data

18. Производительность и оптимизация

# ═══════ Bulk операции ═══════
from sqlalchemy import insert, update, delete

# Массовая вставка (быстрее чем add_all)
with Session(engine) as session:
    session.exec(
        insert(Hero),
        [{"name": f"Hero-{i}", "age": i} for i in range(10000)]
    )
    session.commit()

# Массовое обновление
with Session(engine) as session:
    session.exec(
        update(Hero)
        .where(Hero.age < 18)
        .values(team_id=None)
    )
    session.commit()

# Массовое удаление
with Session(engine) as session:
    session.exec(
        delete(Hero).where(Hero.team_id == None)
    )
    session.commit()

# ═══════ Выборка только нужных колонок ═══════
# Вместо select(Hero) — все колонки
select(Hero.id, Hero.name)  # только id и name

# ═══════ Подсчёт без загрузки ═══════
count = session.exec(
    select(func.count()).select_from(Hero)
).one()

# ═══════ Пагинация по курсору (для больших таблиц) ═══════
def get_heroes_cursor(session: Session, after_id: int = 0, limit: int = 20):
    statement = (
        select(Hero)
        .where(Hero.id > after_id)
        .order_by(Hero.id)
        .limit(limit)
    )
    return session.exec(statement).all()

# ═══════ expire_on_commit ═══════
# Отключить автообновление после коммита (быстрее)
with Session(engine, expire_on_commit=False) as session:
    hero = Hero(name="Fast")
    session.add(hero)
    session.commit()
    # hero.name доступен без refresh

Итоговая шпаргалка

╔══════════════════════════════════════════════════════════════╗
║                    SQLModel — ВСЁ В ОДНОМ                    ║
╠══════════════════════════════════════════════════════════════╣
║                                                              ║
║  МОДЕЛИ:        SQLModel + table=True / table=False          ║
║  ПОЛЯ:          Field(FK, index, unique, ge, le, max_length) ║
║  СВЯЗИ:         Relationship(back_populates, link_model)     ║
║  ЗАПРОСЫ:       select().where().join().order_by().limit()    ║
║  СЕССИЯ:        add / commit / refresh / delete / exec       ║
║  ВАЛИДАЦИЯ:     field_validator / model_validator (Pydantic)  ║
║  МИГРАЦИИ:      Alembic + SQLModel.metadata                  ║
║  ASYNC:         AsyncSession + create_async_engine            ║
║  ТЕСТЫ:         sqlite:// + StaticPool + dependency override  ║
║  КАСКАДЫ:       sa_relationship_kwargs={"cascade": "..."}    ║
║  ИНДЕКСЫ:       Field(index=True) / __table_args__           ║
║  JSON:          sa_column=Column(JSON)                       ║
║  ПРОИЗВОД-ТЬ:  bulk insert/update, selectinload, курсоры    ║
║                                                              ║
║  Если не хватает SQLModel → используй sa_column / sa_*       ║
║  SQLModel = SQLAlchemy + Pydantic (всё из обоих доступно)    ║
║                                                              ║
╚══════════════════════════════════════════════════════════════╝

Это покрывает ~95% всех задач с SQLModel. Оставшиеся 5% — это прямое использование SQLAlchemy Core/ORM, которое SQLModel полностью поддерживает через sa_column, sa_relationship_kwargs и прямой доступ к __table__.

Полный гайд по связям в SQLModel

Оглавление

1. Основы
2. One-to-Many (Один ко многим)
3. Many-to-One (Многие к одному)
4. One-to-One (Один к одному)
5. Many-to-Many (Многие ко многим)
6. Самореферентные связи
7. Каскадные операции
8. Загрузка связей (Lazy / Eager)
9. Связи и Pydantic-схемы (read models)
10. Связи + FastAPI
11. Частые ошибки и подводные камни

1. Основы

Два ключевых инструмента

from sqlmodel import SQLModel, Field, Relationship

Базовая настройка

from sqlmodel import SQLModel, Field, Relationship, Session, create_engine, select
from typing import Optional

engine = create_engine("sqlite:///database.db", echo=True)

def create_db():
    SQLModel.metadata.create_all(engine)

2. One-to-Many (Один ко многим)

Один Team → Много Hero

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)
    name: str = Field(index=True)

    # Внешний ключ — реальная колонка в БД
    team_id: int | None = Field(default=None, foreign_key="team.id")

    # Связь: объект команды
    team: Team | None = Relationship(back_populates="heroes")

Создание записей

with Session(engine) as session:
    # Способ 1: Через FK
    team = Team(name="Avengers")
    session.add(team)
    session.commit()
    session.refresh(team)

    hero = Hero(name="Iron Man", team_id=team.id)
    session.add(hero)
    session.commit()

    # Способ 2: Через relationship (удобнее)
    team2 = Team(name="X-Men")
    hero2 = Hero(name="Wolverine", team=team2)
    session.add(hero2)  # team2 добавится автоматически!
    session.commit()

    # Способ 3: Добавление в список
    team3 = Team(name="Justice League")
    team3.heroes = [
        Hero(name="Batman"),
        Hero(name="Superman"),
    ]
    session.add(team3)
    session.commit()

Чтение связей

with Session(engine) as session:
    # Получить команду героя
    hero = session.exec(select(Hero).where(Hero.name == "Iron Man")).first()
    print(hero.team.name)  # "Avengers" — lazy load

    # Получить героев команды
    team = session.exec(select(Team).where(Team.name == "Avengers")).first()
    for h in team.heroes:  # lazy load
        print(h.name)

3. Many-to-One (Многие к одному)

Это обратная сторона One-to-Many. Технически — то же самое, просто смотрим с другой стороны.

class Country(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

    cities: list["City"] = Relationship(back_populates="country")


class City(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

    country_id: int = Field(foreign_key="country.id")  # обязательная связь

    # Many-to-One: много городов → одна страна
    country: Country = Relationship(back_populates="cities")

Правило: FK всегда на стороне "Many". Кто хранит foreign_key — тот "Many".

4. One-to-One (Один к одному)

SQLModel не имеет специального параметра для One-to-One. Используем Relationship + sa_relationship_kwargs.

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    username: str

    profile: "Profile | None" = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"uselist": False}  # ← ключевой момент!
    )


class Profile(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    bio: str = ""

    user_id: int = Field(foreign_key="user.id", unique=True)  # unique!

    user: User = Relationship(back_populates="profile")

uselist=False — говорит SQLAlchemy возвращать один объект, а не список. unique=True на FK — гарантирует уникальность на уровне БД.

with Session(engine) as session:
    user = User(username="john")
    profile = Profile(bio="Hello!", user=user)
    session.add(profile)
    session.commit()

    session.refresh(user)
    print(user.profile.bio)       # "Hello!"
    print(profile.user.username)  # "john"

5. Many-to-Many (Многие ко многим)

Нужна промежуточная (link/association) таблица.

Вариант A: Простая link-таблица

class HeroMissionLink(SQLModel, table=True):
    """Промежуточная таблица — только два FK"""
    hero_id: int = Field(foreign_key="hero.id", primary_key=True)
    mission_id: int = Field(foreign_key="mission.id", primary_key=True)


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

    missions: list["Mission"] = Relationship(
        back_populates="heroes",
        link_model=HeroMissionLink,  # ← указываем link-модель
    )


class Mission(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str

    heroes: list["Hero"] = Relationship(
        back_populates="missions",
        link_model=HeroMissionLink,
    )

Использование

with Session(engine) as session:
    hero1 = Hero(name="Spider-Man")
    hero2 = Hero(name="Black Widow")

    mission1 = Mission(title="Save the World")
    mission2 = Mission(title="Stealth Op")

    # Добавляем через relationship
    hero1.missions = [mission1, mission2]
    hero2.missions = [mission1]

    session.add(hero1)
    session.add(hero2)
    session.commit()

    # Чтение
    session.refresh(mission1)
    for h in mission1.heroes:
        print(h.name)  # Spider-Man, Black Widow

Вариант B: Link-таблица с дополнительными полями

class Enrollment(SQLModel, table=True):
    """Связь студент-курс с дополнительными данными"""
    student_id: int = Field(foreign_key="student.id", primary_key=True)
    course_id: int = Field(foreign_key="course.id", primary_key=True)

    grade: float | None = None         # ← доп. поле
    enrolled_at: str | None = None     # ← доп. поле


class Student(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

    # Связь напрямую с Course через link_model
    courses: list["Course"] = Relationship(
        back_populates="students",
        link_model=Enrollment,
    )

    # Доступ к самим записям Enrollment
    enrollments: list[Enrollment] = Relationship(back_populates="student")


class Course(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str

    students: list["Student"] = Relationship(
        back_populates="courses",
        link_model=Enrollment,
    )

    enrollments: list[Enrollment] = Relationship(back_populates="course")

Чтобы enrollments работал, добавьте обратные связи в Enrollment:

class Enrollment(SQLModel, table=True):
    student_id: int = Field(foreign_key="student.id", primary_key=True)
    course_id: int = Field(foreign_key="course.id", primary_key=True)
    grade: float | None = None

    student: "Student" = Relationship(back_populates="enrollments")
    course: "Course" = Relationship(back_populates="enrollments")

# Создание с доп. данными
with Session(engine) as session:
    student = Student(name="Alice")
    course = Course(title="Math")
    session.add_all([student, course])
    session.commit()

    enrollment = Enrollment(
        student_id=student.id,
        course_id=course.id,
        grade=95.5,
    )
    session.add(enrollment)
    session.commit()

6. Самореферентные связи

Дерево (parent → children)

class Category(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

    parent_id: int | None = Field(default=None, foreign_key="category.id")

    # Родитель
    parent: "Category | None" = Relationship(
        back_populates="children",
        sa_relationship_kwargs={"remote_side": "Category.id"},
    )

    # Дети
    children: list["Category"] = Relationship(back_populates="parent")

with Session(engine) as session:
    root = Category(name="Electronics")
    phones = Category(name="Phones", parent=root)
    laptops = Category(name="Laptops", parent=root)
    iphone = Category(name="iPhone", parent=phones)

    session.add(root)
    session.commit()

    session.refresh(root)
    for child in root.children:
        print(child.name)  # Phones, Laptops

Подписчики (Many-to-Many на себя)

class FollowLink(SQLModel, table=True):
    follower_id: int = Field(foreign_key="usermodel.id", primary_key=True)
    following_id: int = Field(foreign_key="usermodel.id", primary_key=True)


class UserModel(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    username: str

    # Те, на кого я подписан
    following: list["UserModel"] = Relationship(
        back_populates="followers",
        link_model=FollowLink,
        sa_relationship_kwargs={
            "primaryjoin": "UserModel.id == FollowLink.follower_id",
            "secondaryjoin": "UserModel.id == FollowLink.following_id",
        },
    )

    # Мои подписчики
    followers: list["UserModel"] = Relationship(
        back_populates="following",
        link_model=FollowLink,
        sa_relationship_kwargs={
            "primaryjoin": "UserModel.id == FollowLink.following_id",
            "secondaryjoin": "UserModel.id == FollowLink.follower_id",
        },
    )

7. Каскадные операции

Каскады настраиваются через sa_relationship_kwargs:

class Author(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

    books: list["Book"] = Relationship(
        back_populates="author",
        sa_relationship_kwargs={
            "cascade": "all, delete-orphan",  # ← каскадное удаление
        },
    )


class Book(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    author_id: int = Field(foreign_key="author.id")

    author: Author = Relationship(back_populates="books")

Варианты каскадов

with Session(engine) as session:
    author = Author(name="Tolkien", books=[
        Book(title="The Hobbit"),
        Book(title="LOTR"),
    ])
    session.add(author)
    session.commit()

    # Удаляем автора → книги удалятся автоматически
    session.delete(author)
    session.commit()