# Все важные темы 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. Определение моделей

```python
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. Типы полей и валидация

```python
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

| Python | SQLite | PostgreSQL | MySQL |
|---|---|---|---|
| `int` | INTEGER | INTEGER | INTEGER |
| `float` | FLOAT | FLOAT | FLOAT |
| `str` | VARCHAR | VARCHAR | VARCHAR |
| `bool` | BOOLEAN | BOOLEAN | BOOLEAN |
| `datetime` | DATETIME | TIMESTAMP | DATETIME |
| `date` | DATE | DATE | DATE |
| `time` | TIME | TIME | TIME |
| `Decimal` | NUMERIC | NUMERIC | DECIMAL |
| `bytes` | BLOB | BYTEA | BLOB |
| `uuid.UUID` | CHAR(32) | UUID | CHAR(36) |

---

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

```python
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

```python
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-операции

```python
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. Запросы (самая обширная тема)

```python
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)

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

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

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

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

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

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

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

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

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

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

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

```python
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. Паттерн множественных моделей

```python
# ═══════ 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. Индексы и ограничения

```python
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)

```python
# 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 (продвинутая)

```python
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. События и хуки

```python
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

```python
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 и сложные типы

```python
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),
    )
```

```python
# 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 и гибридные свойства

```python
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. Тестирование

```python
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. Производительность и оптимизация

```python
# ═══════ 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__`.