Шпаргалка по SQLAlchemy: ORM, сессии, запросы, миграции

18 июля 2026 г.от txtly

SQLAlchemy — самый популярный Python ORM. Два слоя: Core (SQL-конструктор, connection pooling) и ORM (модели, сессии, отношения). Шпаргалка по версии 2.0+ — современный API с type hints и select() вместо устаревшего Query.

Быстрый старт

pip install sqlalchemy alembic
# async: pip install sqlalchemy[asyncio] aiosqlite  # или asyncpg для PostgreSQL

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# 1. Базовый класс
class Base(DeclarativeBase):
    pass

# 2. Движок (engine) — управляет пулом соединений
engine = create_engine("sqlite:///app.db", echo=True)  # echo=True — логировать SQL

# 3. Создать таблицы
Base.metadata.create_all(engine)

# 4. Фабрика сессий
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(bind=engine)

# 5. Работа с БД
with SessionLocal() as session:
    session.add(User(name="Анна"))
    session.commit()

Подключение к БД — Engine

Engine управляет пулом соединений. Создаётся один раз при старте приложения.

from sqlalchemy import create_engine

# SQLite (файл)
engine = create_engine("sqlite:///app.db")

# SQLite (в памяти — для тестов)
engine = create_engine("sqlite:///:memory:")

# PostgreSQL
engine = create_engine("postgresql://user:pass@localhost/dbname")

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

# Настройка пула соединений
engine = create_engine(
    "postgresql://user:pass@localhost/db",
    pool_size=20,          # максимальное число соединений
    max_overflow=10,       # дополнительных сверх pool_size
    pool_pre_ping=True,    # проверять соединение перед использованием
    pool_recycle=3600,     # пересоздавать соединения старше 1 часа
    echo=False,            # не логировать SQL в production
)

Модели и типы колонок

Определение модели (SQLAlchemy 2.0+ стиль)

Модель = таблица в БД. Каждый атрибут — колонка. Mapped[тип] — аннотация Python-типа, mapped_column() — конфигурация SQL-типа и ограничений.

from datetime import datetime, date
from decimal import Decimal
from typing import Optional
from sqlalchemy import String, Integer, Float, Numeric, Boolean, DateTime, Date, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    age: Mapped[int] = mapped_column(Integer, default=0)
    balance: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)    # опциональное поле
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, onupdate=func.now())

    def __repr__(self):
        return f"User(id={self.id}, username='{self.username}')"

Ограничения и индексы

Ограничения гарантируют целостность данных на уровне БД — даже если баг в коде, БД не даст записать невалидные данные.

from sqlalchemy import CheckConstraint, UniqueConstraint, Index, ForeignKey

class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)  # уникальность на уровне колонки
    price: Mapped[float] = mapped_column(Float)

    # Табличные ограничения
    __table_args__ = (
        CheckConstraint("price > 0", name="ck_price_positive"),  # цена > 0
        UniqueConstraint("name", "category_id", name="uq_name_category"),  # уникальная пара
        Index("idx_price", "price"),                              # индекс для поиска по цене
        Index("idx_name_price", "name", "price"),                 # составной индекс
    )

Отношения (relationships)

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

Один пользователь → много постов. Самый частый тип связи. FK ставится на стороне «много».

from typing import List
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))

    # Один пользователь → много постов
    posts: Mapped[List["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan"      # удалить посты при удалении пользователя
    )

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))  # внешний ключ

    # Много постов → один пользователь
    author: Mapped["User"] = relationship(back_populates="posts")

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

Пост может иметь много тегов, тег — много постов. Нужна промежуточная таблица.

from sqlalchemy import Table, Column

# Промежуточная таблица
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))

    tags: Mapped[List["Tag"]] = relationship(secondary=post_tags, back_populates="posts")

class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))

    posts: Mapped[List["Post"]] = relationship(secondary=post_tags, back_populates="tags")

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

Пользователь ↔ Профиль. FK с unique=True или uselist=False.

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    profile: Mapped["Profile"] = relationship(back_populates="user", uselist=False)

class Profile(Base):
    __tablename__ = "profiles"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), unique=True)
    bio: Mapped[str] = mapped_column(Text)
    user: Mapped["User"] = relationship(back_populates="profile")

Self-referential (ссылка на себя)

Категория → родительская категория.

class Category(Base):
    __tablename__ = "categories"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    parent_id: Mapped[Optional[int]] = mapped_column(ForeignKey("categories.id"))

    parent: Mapped[Optional["Category"]] = relationship(
        "Category", remote_side=[id], back_populates="children"
    )
    children: Mapped[List["Category"]] = relationship(
        "Category", back_populates="parent"
    )

CRUD — создание, чтение, обновление, удаление

Create — создание записей

with SessionLocal() as session:
    # Одна запись
    user = User(username="anna", email="anna@mail.com")
    session.add(user)
    session.commit()
    session.refresh(user)              # получить id, created_at из БД

    # Несколько записей
    users = [User(username=f"user_{i}", email=f"user{i}@mail.com") for i in range(10)]
    session.add_all(users)
    session.commit()

    # Массовая вставка (быстрее для тысяч записей — не создаёт Python-объекты)
    from sqlalchemy import insert
    session.execute(
        insert(User).values([{"username": f"u{i}", "email": f"u{i}@m.com"} for i in range(1000)])
    )
    session.commit()

Read — чтение записей

Современный подход — select() + session.scalars(), НЕ session.query() (старый API).

from sqlalchemy import select

with SessionLocal() as session:
    # По primary key
    user = session.get(User, 1)          # быстро, использует кеш сессии

    # Одна запись по условию
    user = session.scalar(select(User).where(User.username == "anna"))

    # Все записи
    users = session.scalars(select(User)).all()

    # Фильтрация
    stmt = select(User).where(User.age >= 18, User.is_active == True)
    active_adults = session.scalars(stmt).all()

    # LIKE, IN, BETWEEN
    stmt = select(User).where(User.email.like("%@gmail.com"))       # LIKE
    stmt = select(User).where(User.id.in_([1, 2, 3]))               # IN
    stmt = select(User).where(User.age.between(18, 65))             # BETWEEN

    # Сортировка и пагинация
    stmt = select(User).order_by(User.created_at.desc()).limit(10).offset(20)
    page = session.scalars(stmt).all()

    # Только нужные колонки
    stmt = select(User.id, User.username).where(User.is_active == True)
    for row in session.execute(stmt):
        print(row.id, row.username)

Update — обновление

with SessionLocal() as session:
    # Обновление объекта
    user = session.get(User, 1)
    user.age = 31
    session.commit()

    # Массовое обновление (быстрее — не загружает объекты)
    from sqlalchemy import update
    stmt = (update(User)
            .where(User.is_active == False)
            .values(is_deleted=True, deleted_at=func.now()))
    session.execute(stmt)
    session.commit()

Delete — удаление

with SessionLocal() as session:
    # Удаление объекта
    user = session.get(User, 42)
    if user:
        session.delete(user)
        session.commit()

    # Массовое удаление
    from sqlalchemy import delete
    stmt = delete(User).where(User.is_active == False)
    session.execute(stmt)
    session.commit()

Запросы — joins, агрегации, подзапросы

Joins и eager loading

N+1 проблема: загрузили 10 пользователей, для каждого пошли в БД за постами = 11 запросов вместо 1-2. Eager loading решает это.

from sqlalchemy.orm import selectinload, joinedload, contains_eager

# selectinload — отдельный SELECT IN для связанных объектов
# Лучше для one-to-many (не дублирует строки)
stmt = select(User).options(selectinload(User.posts))
users = session.scalars(stmt).unique().all()

# joinedload — LEFT JOIN в одном запросе
# Лучше для many-to-one (одна связанная запись)
stmt = select(Post).options(joinedload(Post.author))

# Вложенная загрузка
stmt = select(User).options(
    selectinload(User.posts).selectinload(Post.comments)
)

# Явный JOIN
stmt = (select(User, Post)
        .join(Post, User.id == Post.user_id)
        .where(Post.published == True))

# LEFT JOIN — все пользователи, даже без постов
stmt = select(User).outerjoin(User.posts)

# contains_eager — использовать данные из явного JOIN
stmt = (select(User)
        .join(User.posts)
        .options(contains_eager(User.posts))
        .where(Post.published == True))

Агрегации

from sqlalchemy import func, desc

# COUNT, AVG, SUM, MIN, MAX
total = session.scalar(select(func.count(User.id)))

avg_age = session.scalar(select(func.avg(User.age)))

# GROUP BY + HAVING
stmt = (select(User.city, func.count(User.id).label("cnt"))
        .group_by(User.city)
        .having(func.count(User.id) > 10)
        .order_by(desc("cnt")))

Подзапросы и CTE

# Подзапрос
subq = select(func.avg(User.age)).scalar_subquery()
stmt = select(User).where(User.age > subq)

# CTE (Common Table Expression)
cte = select(User.city, func.count().label("cnt")).group_by(User.city).cte()
stmt = select(cte).where(cte.c.cnt > 10)

# Оконные функции
from sqlalchemy import func
stmt = select(
    User.name,
    User.salary,
    func.avg(User.salary).over(partition_by=User.department).label("dept_avg"),
    func.rank().over(order_by=User.salary.desc()).label("rank"),
)

Сырой SQL

from sqlalchemy import text

# Параметризованный запрос
stmt = text("SELECT * FROM users WHERE age > :age")
result = session.execute(stmt, {"age": 18})

# Смешанный: ORM + сырой SQL
stmt = select(User).where(text("extract(year from created_at) = :year")).params(year=2026)

Сессии и транзакции

Сессия — единица работы с БД. Отслеживает изменения объектов и синхронизирует их с БД при commit().

from sqlalchemy.orm import Session, sessionmaker

SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,        # ручное управление транзакциями (всегда так)
    autoflush=True,          # auto-flush перед запросами (удобно)
    expire_on_commit=False,  # не сбрасывать объекты после commit (удобно для FastAPI)
)

# Контекстный менеджер — рекомендуемый способ
with SessionLocal() as session:
    user = session.get(User, 1)
    user.age += 1
    session.commit()         # или session.rollback() при ошибке

# Ручное управление транзакцией
with SessionLocal() as session:
    with session.begin():    # commit в конце блока, rollback при исключении
        user = User(name="Анна")
        session.add(user)

# rollback вручную
session = SessionLocal()
try:
    user = User(name="Анна")
    session.add(user)
    session.commit()
except Exception:
    session.rollback()
    raise
finally:
    session.close()

Асинхронные сессии

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///app.db")
# engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_user(user_id: int):
    async with AsyncSessionLocal() as session:
        user = await session.get(User, user_id)
        return user

async def create_user(name: str):
    async with AsyncSessionLocal() as session:
        async with session.begin():
            user = User(name=name)
            session.add(user)
        # commit автоматический
        return user

Alembic — миграции

Alembic управляет схемой БД в коде — как Git для структуры базы данных. Каждое изменение схемы = файл миграции, который можно применить, откатить и положить в репозиторий.

pip install alembic
alembic init alembic          # создать папку alembic/

В alembic/env.py указать целевую модель:

from myapp.models import Base   # ваш Base с DeclarativeBase
target_metadata = Base.metadata

Команды

alembic revision --autogenerate -m "Add users table"   # создать миграцию по моделям
alembic upgrade head                 # применить все миграции
alembic upgrade +1                   # применить одну
alembic downgrade -1                 # откатить одну
alembic downgrade base               # откатить все
alembic current                      # текущая ревизия
alembic history                      # история миграций
alembic check                        # проверить, не отстала ли БД от моделей

Пример файла миграции

"""Add users table

Revision ID: abc123
Revises: def456
"""
from alembic import op
import sqlalchemy as sa

revision = 'abc123'
down_revision = 'def456'

def upgrade():
    op.create_table('users',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('username', sa.String(50), nullable=False),
        sa.Column('email', sa.String(100)),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
    )
    op.create_index('ix_users_username', 'users', ['username'])

def downgrade():
    op.drop_index('ix_users_username', 'users')
    op.drop_table('users')

Миграция с данными

from alembic import op
from sqlalchemy import text

def upgrade():
    # Схема
    op.add_column('users', sa.Column('full_name', sa.String(200)))

    # Данные
    conn = op.get_bind()
    conn.execute(text("UPDATE users SET full_name = first_name || ' ' || last_name"))

    # Удалить старые колонки (если нужно)
    op.drop_column('users', 'first_name')
    op.drop_column('users', 'last_name')

def downgrade():
    op.add_column('users', sa.Column('first_name', sa.String(100)))
    op.add_column('users', sa.Column('last_name', sa.String(100)))
    conn = op.get_bind()
    conn.execute(text("UPDATE users SET first_name = split_part(full_name, ' ', 1)"))
    conn.execute(text("UPDATE users SET last_name = split_part(full_name, ' ', 2)"))
    op.drop_column('users', 'full_name')

Приёмы и best practices

Динамический построитель запросов

def search_users(session: Session, **filters):
    stmt = select(User)
    if name := filters.get("name"):
        stmt = stmt.where(User.name.ilike(f"%{name}%"))
    if min_age := filters.get("min_age"):
        stmt = stmt.where(User.age >= min_age)
    if is_active := filters.get("is_active"):
        stmt = stmt.where(User.is_active == is_active)
    return session.scalars(stmt).all()

Обработка ошибок

from sqlalchemy.exc import IntegrityError, NoResultFound

def create_user(session: Session, username: str, email: str):
    try:
        user = User(username=username, email=email)
        session.add(user)
        session.commit()
        return user
    except IntegrityError:
        session.rollback()
        raise ValueError("Пользователь с таким именем или email уже существует")

def get_user_or_404(session: Session, user_id: int):
    user = session.get(User, user_id)
    if not user:
        raise ValueError(f"User {user_id} not found")
    return user

Тестирование с in-memory SQLite

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, User

@pytest.fixture
def db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    try:
        yield session
    finally:
        session.close()

def test_create_user(db_session):
    user = User(username="test", email="test@mail.com")
    db_session.add(user)
    db_session.commit()
    assert user.id is not None

Стратегии ленивой загрузки

class User(Base):
    posts: Mapped[List["Post"]] = relationship(
        lazy="selectin",           # всегда загружать через SELECT IN
        # "joined"  — через JOIN (один запрос, но дублируются строки пользователя)
        # "select"  — лениво (по умолчанию, N+1 проблема)
        # "raise"   — ошибка при обращении (защита от N+1)
        # "noload"  — не загружать, вернуть пустой список
        # "immediate" — загрузить сразу после загрузки родителя
        # "subquery" — через подзапрос (устаревший)
        cascade="all, delete-orphan",
    )

Частые ошибки

Ошибка	Причина	Решение
`MissingGreenlet` в async	Синхронная сессия в async-контексте	`AsyncSession` + `async_sessionmaker`
N+1 запросов	Ленивая загрузка `lazy="select"`	`selectinload(User.posts)`
Объекты «пустые» после commit	`expire_on_commit=True` (по умолчанию)	`expire_on_commit=False` или `refresh()`
`DetachedInstanceError`	Использование объекта вне сессии	Загружать связанные объекты внутри сессии
`IntegrityError` UNIQUE constraint	Дубликат unique-поля	Проверять перед вставкой или обрабатывать исключение
`sqlalchemy.exc.ArgumentError`	Разные Base-классы в разных файлах	Один `DeclarativeBase` на всё приложение
Миграции не создаются	Alembic не видит модели	Проверить `target_metadata` в `env.py`
FK constraint failed	Ссылка на несуществующую запись	Создать родительскую запись перед дочерней

Ресурсы

Что дальше

Связанные шпаргалки: Python, FastAPI, Django, SQL

Языки программирования Шпаргалки

sqlalchemy python шпаргалка