31 вопрос по теме «ORM и SQLAlchemy» на собеседовании

Короткий ответ: ORM (Object-Relational Mapping) — слой, отображающий строки реляционных таблиц на объекты языка программирования. Вы работаете с классами и атрибутами, а ORM генерирует SQL. Плюсы — продуктивность и безопасность, минусы — потеря контроля над SQL и риск скрытых тормозов.

Подробно:

# SQLAlchemy
class User(Base):
    __tablename__ = "users"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String)

user = session.get(User, 1)   # SELECT * FROM users WHERE id = 1
user.name = "Alice"           # объект в памяти
session.commit()              # UPDATE users SET name='Alice' WHERE id=1

# Django ORM
class User(models.Model):
    name = models.CharField(max_length=100)

user = User.objects.get(id=1)
user.name = "Alice"
user.save()

Плюсы:

• Меньше шаблонного кода, нет ручного маппинга строк в объекты.
• Параметризованные запросы из коробки → защита от SQL-инъекций.
• Переносимость между СУБД (PostgreSQL, MySQL, SQLite).
• Миграции, валидация, связи, кэш в рамках сессии.

Минусы:

• Скрывает SQL → легко породить N+1 и неэффективные запросы.
• «Дырявая абстракция»: сложные запросы всё равно требуют знания SQL.
• Дополнительный слой и накладные расходы.
• Иногда генерирует неоптимальный SQL.

⚠️ Ловушка: ORM не освобождает от знания SQL. Чтобы диагностировать тормоза, всё равно нужно смотреть на сгенерированный SQL и план запроса (EXPLAIN).

Короткий ответ: ORM экономит время на 90% типовых CRUD-запросов и даёт безопасность/единообразие. Сырой SQL пишут там, где ORM мешает: сложная аналитика, оконные функции, рекурсивные CTE, тонкая оптимизация, bulk-операции.

Подробно: ORM закрывает рутину (вставки, выборки по ключу, простые join-ы, связи). Но он — абстракция, и на сложных запросах либо генерирует плохой SQL, либо не умеет конструкцию вовсе.

Признаки, что пора писать SQL:

• Оконные функции, рекурсивные CTE, сложные GROUP BY ... HAVING.
• Массовые UPDATE ... FROM / INSERT ... SELECT.
• Запрос на горячем пути, где важна каждая миллисекунда.
• Сложный отчёт, который в ORM выглядит как нечитаемый набор annotate.

# SQLAlchemy — сырой SQL, но с параметрами (безопасно)
from sqlalchemy import text
rows = session.execute(
    text("SELECT id, name FROM users WHERE age > :age"),
    {"age": 18},
).all()

# Django — raw() возвращает модели
users = User.objects.raw("SELECT * FROM users WHERE age > %s", [18])
# Или connection.cursor() для произвольного SQL
from django.db import connection
with connection.cursor() as cur:
    cur.execute("SELECT count(*) FROM users WHERE age > %s", [18])
    total = cur.fetchone()[0]

⚠️ Ловушка: «сырой SQL» не значит «строковая конкатенация». Всегда используйте параметры (:age, %s), иначе получите SQL-инъекцию.

Короткий ответ: Это фундаментальное несоответствие между объектной моделью (наследование, ссылки, инкапсуляция, идентичность по ссылке) и реляционной (таблицы, строки, внешние ключи, идентичность по ключу). ORM пытается сгладить это несоответствие, но не устраняет полностью.

Подробно: Основные точки расхождения:

• Идентичность: в ОО — по ссылке/identity, в БД — по первичному ключу. ORM решает через identity map.
• Наследование: в ОО есть, в реляционной модели — нет. ORM эмулирует (single table / joined table / concrete inheritance).
• Связи: объект ссылается на объект напрямую; в БД — через FK и JOIN. Отсюда ленивая загрузка и N+1.
• Гранулярность: объект-значение (Address) vs набор колонок.
• Навигация: в коде — user.orders[0].items, в SQL — это уже несколько JOIN/запросов.

⚠️ Ловушка: «навигация по объектам бесплатна» — иллюзия. Каждый переход по связи может быть отдельным SQL-запросом.

Короткий ответ: N+1 — это когда для получения списка из N объектов и их связанных данных выполняется 1 запрос на список + N запросов (по одному на связь каждого объекта). Возникает из-за ленивой загрузки. Решается eager loading (JOIN или batched-запрос).

Подробно: ORM провоцирует её, потому что навигация по связи (user.orders) выглядит как обращение к атрибуту, а под капотом — отдельный SQL.

# Django — N+1 (1 запрос на users + по 1 на каждый user.profile)
for user in User.objects.all():        # 1 запрос
    print(user.profile.bio)            # +N запросов

# Решение
for user in User.objects.select_related("profile"):  # 1 запрос с JOIN
    print(user.profile.bio)

# SQLAlchemy — N+1
for user in session.query(User).all():     # 1 запрос
    print(user.profile.bio)                # +N запросов (lazy)

# Решение
from sqlalchemy.orm import selectinload
users = session.scalars(
    select(User).options(selectinload(User.orders))
).all()  # 2 запроса вместо N+1

Как обнаружить:

• Включить логирование SQL (echo=True в SQLAlchemy, django-debug-toolbar, логгер django.db.backends).
• Видите много одинаковых SELECT ... WHERE id = ? в цикле — это N+1.

⚠️ Ловушка: eager loading «на всякий случай» для всех связей тоже вредит — лишние JOIN и большие объёмы данных. Грузите только то, что реально используете.

Короткий ответ: Lazy (ленивая) — связанные данные грузятся в момент первого обращения, отдельным запросом. Eager (жадная) — связанные данные грузятся сразу вместе с основным объектом. Lazy экономит память когда связь не нужна, но провоцирует N+1; eager экономит количество запросов.

Подробно:

# SQLAlchemy: стратегия задаётся в relationship или через options
class User(Base):
    orders = relationship("Order", lazy="select")    # по умолчанию — lazy
    # lazy="selectin" — eager отдельным IN-запросом
    # lazy="joined"   — eager через JOIN
    # lazy="raise"    — бросить ошибку при попытке ленивой загрузки (защита от N+1!)

Значения lazy в SQLAlchemy:

• select (по умолчанию) — отдельный SELECT при обращении.
• selectin — отдельный запрос WHERE id IN (...) для всей партии.
• joined — LEFT OUTER JOIN сразу.
• subquery — подзапрос.
• raise / raise_on_sql — запретить ленивую загрузку (помогает ловить N+1 в тестах).
• noload — не грузить вообще.

⚠️ Ловушка: lazy="raise" на проде/в тестах — отличный способ заставить N+1 падать явно, а не молча тормозить.

Короткий ответ: select_related делает SQL JOIN и подходит для связей «к одному» (ForeignKey, OneToOne). prefetch_related делает отдельный запрос и Python-склейку, подходит для «ко многим» (reverse FK, ManyToMany).

Подробно:

# select_related — один запрос с JOIN, для FK / OneToOne (объект-владелец ссылки)
orders = Order.objects.select_related("customer", "customer__city")
# SELECT ... FROM orders JOIN customers ... JOIN cities ...

# prefetch_related — отдельный запрос + склейка в Python, для M2M / reverse FK
authors = Author.objects.prefetch_related("books")
# SELECT * FROM authors;
# SELECT * FROM books WHERE author_id IN (...);

# Тонкая настройка prefetch с фильтрацией/сортировкой
from django.db.models import Prefetch
authors = Author.objects.prefetch_related(
    Prefetch("books", queryset=Book.objects.filter(published=True))
)

⚠️ Ловушка: select_related нельзя использовать для ManyToMany — JOIN породил бы декартово размножение строк. Для «ко многим» только prefetch_related.

Короткий ответ: Это стратегии eager loading. joinedload — один запрос с JOIN (хорош для «к одному»). selectinload — второй запрос WHERE pk IN (...) (лучший выбор для коллекций). subqueryload — второй запрос с подзапросом (устаревший по сравнению с selectinload).

Подробно:

from sqlalchemy.orm import joinedload, selectinload, subqueryload

# joinedload — LEFT OUTER JOIN, всё в одном запросе
session.scalars(select(User).options(joinedload(User.address)))

# selectinload — основной запрос + SELECT ... WHERE user_id IN (1,2,3,...)
session.scalars(select(User).options(selectinload(User.orders)))

# subqueryload — основной запрос + запрос с подзапросом (повторяет JOIN/ORDER)
session.scalars(select(User).options(subqueryload(User.orders)))

Когда что:

• joinedload — для отношений «к одному» (many-to-one, one-to-one). Для коллекций даёт дублирование строк.
• selectinload — лучший дефолт для коллекций (one-to-many, many-to-many): нет дублирования, эффективный IN.
• subqueryload — историческая альтернатива selectinload; обычно selectinload лучше (особенно при пагинации с LIMIT).

⚠️ Ловушка: joinedload коллекции вместе с LIMIT ломает пагинацию — JOIN размножает строки, и LIMIT отсекает «не те» строки. Для коллекций с лимитом используйте selectinload.

Короткий ответ: Core — это нижний уровень: SQL Expression Language и работа с таблицами/запросами без классов-моделей. ORM — верхний уровень поверх Core с маппингом классов и Session. Engine управляет пулом соединений; Session — рабочая «единица работы» для ORM.

Подробно:

from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import Session

# Engine — фабрика соединений + пул. Создаётся ОДИН раз на приложение.
engine = create_engine("postgresql+psycopg://u:p@host/db", pool_size=5, max_overflow=10)

# Core — без моделей
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users"))

# ORM — Session поверх Engine
with Session(engine) as session:
    user = session.scalars(select(User).where(User.id == 1)).one()

• Engine — глобальный объект, владеет пулом соединений (Connection Pool). Создаётся один раз.
• Connection pool — переиспользует TCP-соединения с БД, чтобы не открывать новое на каждый запрос.
• Session — кратковременный объект: identity map + unit of work + транзакция. Создаётся на запрос/операцию.

⚠️ Ловушка: Engine — долгоживущий и потокобезопасный (один на приложение). Session — короткоживущий и НЕ потокобезопасный (один на запрос/поток). Их жизненные циклы путать нельзя.

Короткий ответ: Declarative — способ описывать модели классами, наследуясь от Base. relationship задаёт связь между моделями. back_populates явно указывает парный атрибут на другой стороне; backref создаёт его автоматически. Сейчас рекомендуют back_populates за явность.

Подробно:

from sqlalchemy.orm import DeclarativeBase, relationship, mapped_column
from sqlalchemy import ForeignKey

class Base(DeclarativeBase): ...

class User(Base):
    __tablename__ = "users"
    id = mapped_column(Integer, primary_key=True)
    # явная двусторонняя связь
    orders = relationship("Order", back_populates="user")

class Order(Base):
    __tablename__ = "orders"
    id = mapped_column(Integer, primary_key=True)
    user_id = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="orders")

# backref — короче, генерирует Order.user автоматически:
# orders = relationship("Order", backref="user")

⚠️ Ловушка: при back_populates нужно объявить связь на ОБЕИХ сторонах с симметричными именами, иначе обновление одной стороны не отразится на другой в памяти.

Короткий ответ: add ставит объект в очередь сессии (pending), flush отправляет SQL в БД в рамках текущей транзакции (но не фиксирует), commit фиксирует транзакцию, rollback откатывает, expire помечает атрибуты устаревшими — при следующем обращении они перечитаются из БД.

Подробно:

session.add(user)        # объект -> pending
session.flush()          # INSERT в БД, но транзакция ещё открыта; user.id уже доступен
session.commit()         # COMMIT; по умолчанию объекты становятся expired
session.rollback()       # ROLLBACK; pending-изменения отброшены
session.expire(user)     # сбросить загруженные атрибуты -> перечитаются по требованию
session.refresh(user)    # немедленно перечитать из БД

Состояния объекта: transient → pending (после add) → persistent (после flush/commit) → detached (после закрытия сессии) / deleted.

⚠️ Ловушка: после commit() по умолчанию (expire_on_commit=True) все атрибуты объектов помечаются expired. Обращение к ним вне сессии (например, при сериализации после закрытия) выбросит DetachedInstanceError. Решения: expire_on_commit=False или загрузить нужные данные до закрытия.

Короткий ответ: flush отправляет накопленные изменения (INSERT/UPDATE/DELETE) в БД в рамках открытой транзакции, но НЕ фиксирует их — они ещё могут быть откачены. commit завершает транзакцию (внутри сначала делает flush, затем COMMIT) — изменения становятся постоянными и видимыми другим транзакциям.

Подробно:

session.add(user)
session.flush()       # SQL ушёл в БД; user.id присвоен; видно в ЭТОЙ транзакции
print(user.id)        # доступно
session.rollback()    # всё откатилось — пользователь НЕ сохранён

session.add(user2)
session.commit()      # flush + COMMIT — сохранено навсегда

Зачем нужен отдельный flush:

• Получить автогенерируемый PK до commit (для вставки связанных строк).
• Проверить ограничения БД (unique, FK) в середине транзакции.
• Сохранить семантику: несколько flush, один commit на бизнес-операцию.

⚠️ Ловушка: flush ≠ сохранение. Если после flush упадёт исключение и произойдёт rollback, данные не попадут в БД. Постоянными они становятся только после commit.

Короткий ответ: Identity map — кэш в рамках сессии: каждая строка БД с данным PK представлена ровно одним Python-объектом. Unit of Work — паттерн, при котором сессия накапливает все изменения и применяет их одним согласованным набором SQL при flush/commit.

Подробно:

a = session.get(User, 1)
b = session.get(User, 1)
assert a is b          # тот же объект — identity map, второй SELECT не выполнялся

# Unit of Work: меняем несколько объектов, ORM сам решает порядок INSERT/UPDATE/DELETE
u.name = "X"
session.add(Order(user=u))
session.commit()       # один согласованный набор SQL с учётом зависимостей FK

Зачем:

• Identity map устраняет дубли объектов и лишние запросы в рамках сессии.
• Unit of Work упорядочивает операции (сначала родитель, потом ребёнок) и собирает их в одну транзакцию.

⚠️ Ловушка: identity map живёт в пределах одной сессии. Два запроса в разных сессиях вернут РАЗНЫЕ объекты для одной строки — и один может содержать устаревшие данные (stale).

Короткий ответ: autoflush (включён по умолчанию) автоматически делает flush перед выполнением запроса, чтобы запрос «видел» ещё не сохранённые изменения текущей сессии. Это удобно, но иногда даёт неожиданные ранние INSERT/UPDATE.

Подробно:

session.add(User(name="new"))
# запрос ниже увидит "new", т.к. перед SELECT произойдёт autoflush
users = session.scalars(select(User)).all()

# отключить временно
with session.no_autoflush:
    ... # запросы не будут триггерить flush

⚠️ Ловушка: autoflush может «выстрелить» INSERT раньше, чем вы дозаполнили обязательные поля объекта, → ошибка NOT NULL/constraint в неожиданном месте. В таких случаях оборачивайте код в session.no_autoflush.

Короткий ответ: Session НЕ потокобезопасна — её нельзя шарить между потоками/запросами. Паттерн «session per request» — создавать отдельную сессию на каждый веб-запрос и закрывать в конце. scoped_session даёт по сессии на поток/контекст автоматически.

Подробно:

from sqlalchemy.orm import sessionmaker, scoped_session

SessionFactory = sessionmaker(bind=engine)

# scoped_session — реестр сессий по потоку (thread-local)
Session = scoped_session(SessionFactory)

def handle_request():
    try:
        do_work(Session)
        Session.commit()
    except Exception:
        Session.rollback()
        raise
    finally:
        Session.remove()   # вернуть/закрыть сессию в конце запроса

В async-приложениях используют async_scoped_session со scopefunc на контекст задачи. В FastAPI обычно делают сессию через зависимость (dependency) на запрос.

⚠️ Ловушка: глобальная одна сессия на всё приложение — классическая ошибка. Конкурентные запросы перетрут друг другу состояние и identity map. Сессия должна жить ровно один запрос/задачу.

Короткий ответ: QuerySet ленив — он не обращается к БД при создании и при цепочке фильтров. Запрос выполняется только при «материализации» (итерация, list(), len(), индексация, bool(), срез с шагом). Результат кэшируется в самом QuerySet — повторная итерация по тому же объекту не делает новый запрос.

Подробно:

qs = User.objects.filter(active=True)   # SQL НЕ выполнен
qs = qs.exclude(banned=True)            # всё ещё нет SQL

for u in qs:        # ВОТ ТУТ выполняется SELECT и кэшируется
    ...
for u in qs:        # из кэша, нового запроса нет

# Но новый QuerySet (другой объект) — новый запрос:
list(User.objects.filter(active=True))  # запрос 1
list(User.objects.filter(active=True))  # запрос 2 (другой объект)

Что триггерит выполнение: for, list(), len(), bool(), if qs:, qs[2], list(qs[1:5:2]), repr().

⚠️ Ловушка: if qs.exists() дешевле, чем if qs: или if len(qs) — последние материализуют весь результат. И кэш привязан к конкретному объекту QuerySet, а не к запросу: пересоздание QuerySet = новый запрос.

Короткий ответ: filter/exclude — WHERE; annotate — добавить вычисляемое поле к каждой строке (GROUP BY при агрегатах); aggregate — свернуть весь queryset в словарь; values/values_list — вернуть dict/tuple вместо моделей; only/defer — выбрать/исключить конкретные колонки.

Подробно:

from django.db.models import Count, Sum, Avg

User.objects.filter(age__gte=18).exclude(banned=True)

# annotate — на каждую группу
Author.objects.annotate(num_books=Count("books"))   # ... GROUP BY author

# aggregate — на весь набор, возвращает dict
Order.objects.aggregate(total=Sum("amount"), avg=Avg("amount"))
# {'total': 1000, 'avg': 50}

# values / values_list — без создания объектов моделей (легче)
User.objects.values("id", "name")            # [{'id':1,'name':'A'}, ...]
User.objects.values_list("id", flat=True)    # [1, 2, 3]

# only / defer — управление загружаемыми колонками
User.objects.only("id", "name")    # SELECT только id, name (остальное — lazy)
User.objects.defer("bio")          # SELECT всё, кроме bio

⚠️ Ловушка: only()/defer() грузят отложенные поля лениво — обращение к отложенному полю в цикле снова порождает N+1. И annotate(Count(...)) неявно добавляет GROUP BY, что может изменить число строк, если есть JOIN.

Короткий ответ: F() ссылается на значение поля БД в выражении — операция выполняется на стороне БД атомарно, без гонок. Q() позволяет строить сложные условия с OR/NOT и комбинировать их логически.

Подробно:

from django.db.models import F, Q

# F — обновление на стороне БД, без race condition
Product.objects.filter(id=1).update(stock=F("stock") - 1)
# UPDATE products SET stock = stock - 1 WHERE id = 1   (атомарно)

# сравнение полей между собой
Order.objects.filter(shipped__lt=F("deadline"))

# Q — OR / NOT / сложная логика
User.objects.filter(Q(age__lt=18) | Q(is_staff=True))
User.objects.filter(~Q(status="banned") & Q(active=True))

⚠️ Ловушка: без F() инкремент obj.stock -= 1; obj.save() читает значение в Python и перезаписывает — два параллельных процесса потеряют одно списание (lost update). F() решает это, делая вычисление в БД.

Короткий ответ: bulk_create вставляет много объектов одним (или несколькими) INSERT вместо N отдельных. bulk_update обновляет много объектов пакетно. Они резко быстрее, но обходят часть логики: не вызывают save(), сигналы, и (в части случаев) не возвращают PK.

Подробно:

# Вместо N INSERT — один batch
User.objects.bulk_create(
    [User(name=f"u{i}") for i in range(1000)],
    batch_size=500,
)

# Пакетное обновление
users = list(User.objects.all())
for u in users:
    u.active = False
User.objects.bulk_update(users, ["active"], batch_size=500)

В SQLAlchemy аналог — session.execute(insert(User), [{...}, {...}]) или add_all + один flush.

⚠️ Ловушка: bulk_create/bulk_update НЕ вызывают Model.save(), pre_save/post_save сигналы и auto_now. Если в save() была бизнес-логика — она будет пропущена. На некоторых БД bulk_create не заполняет pk у объектов.

Короткий ответ: В Django транзакции оформляют transaction.atomic() (блок или декоратор) — при выходе без исключения COMMIT, при исключении ROLLBACK. В SQLAlchemy транзакция привязана к Session: session.commit() фиксирует, session.rollback() откатывает; session.begin() даёт явный блок.

Подробно:

# Django
from django.db import transaction

with transaction.atomic():
    order.save()
    payment.save()
    # исключение здесь -> весь блок откатится

@transaction.atomic
def create_order(...):
    ...

# SQLAlchemy 2.0 — явный блок
with Session(engine) as session:
    with session.begin():        # commit на выходе, rollback при ошибке
        session.add(order)
        session.add(payment)
# или вручную: session.add(...); session.commit() / session.rollback()

Когда происходит commit: Django по умолчанию в режиме autocommit оборачивает каждый запрос; внутри atomic — один commit на выходе из самого внешнего блока. SQLAlchemy открывает транзакцию при первом SQL и держит до commit()/rollback().

⚠️ Ловушка: в Django, если поймать исключение ВНУТРИ atomic и не пробросить, транзакция всё равно помечена «битой» — последующие запросы кинут TransactionManagementError. Для частичного отката используйте вложенный atomic (savepoint).

Короткий ответ: Реальных вложенных транзакций в БД нет — вместо них используются SAVEPOINT. Вложенный atomic в Django и session.begin_nested() в SQLAlchemy создают savepoint: их откат отменяет только часть работы, не трогая внешнюю транзакцию.

Подробно:

# Django — внутренний atomic = SAVEPOINT
with transaction.atomic():           # внешняя транзакция
    a.save()
    try:
        with transaction.atomic():   # SAVEPOINT
            b.save()
            raise ValueError
    except ValueError:
        pass                         # откатился только b, a остался
    c.save()
# COMMIT: a и c сохранены, b — нет

# SQLAlchemy — savepoint
with session.begin():
    session.add(a)
    sp = session.begin_nested()      # SAVEPOINT
    try:
        session.add(b)
        session.flush()
        raise ValueError
    except ValueError:
        sp.rollback()                # откат до savepoint
    session.add(c)
# commit: a, c сохранены

⚠️ Ловушка: savepoint удерживает блокировки внешней транзакции. Долгие вложенные транзакции в коде = долгие блокировки строк в БД → дедлоки и просадка конкуренции.

Короткий ответ: Обе системы версионируют изменения схемы. Django генерирует миграции из изменений в моделях (makemigrations) и применяет (migrate); хранит граф зависимостей. Alembic — отдельный инструмент для SQLAlchemy: autogenerate сравнивает модели с БД и пишет ревизию с upgrade()/downgrade().

Подробно:

# Django
python manage.py makemigrations   # сгенерировать из изменений моделей
python manage.py migrate          # применить
python manage.py sqlmigrate app 0002   # посмотреть SQL

# Alembic
alembic revision --autogenerate -m "add users.age"
alembic upgrade head
alembic downgrade -1

# Alembic-ревизия
def upgrade():
    op.add_column("users", sa.Column("age", sa.Integer(), nullable=True))
def downgrade():
    op.drop_column("users", "age")

⚠️ Ловушка: autogenerate в Alembic видит не всё (изменения CHECK, типы колонок, серверные дефолты иногда пропускает). Сгенерированную миграцию ВСЕГДА нужно вычитывать вручную.

Короткий ответ: Опасны блокирующие миграции на больших таблицах и несовместимые со старым кодом изменения. Добавление NOT NULL-колонки без дефолта на заполненной таблице ломает вставки/может блокировать. Решение — многошаговые expand/contract миграции и отдельные data migrations.

Подробно:

Добавление NOT NULL-колонки безопасно в три шага (expand/contract):

# Шаг 1: добавить nullable-колонку
op.add_column("users", sa.Column("status", sa.String(), nullable=True))
# Шаг 2 (data migration): заполнить значения
op.execute("UPDATE users SET status = 'active' WHERE status IS NULL")
# Шаг 3 (после деплоя кода, умеющего писать status): сделать NOT NULL
op.alter_column("users", "status", nullable=False)

# Django data migration
def fill_status(apps, schema_editor):
    User = apps.get_model("app", "User")
    User.objects.filter(status__isnull=True).update(status="active")

class Migration(migrations.Migration):
    operations = [migrations.RunPython(fill_status, migrations.RunPython.noop)]

Принципы zero-downtime:

• Обратная совместимость: новая схема должна работать со старым кодом во время раскатки (rolling deploy).
• Не переименовывать колонки одним шагом — добавить новую, скопировать, переключить код, удалить старую.
• На PostgreSQL ALTER TABLE ... SET NOT NULL/добавление индекса может брать блокировку — используйте CREATE INDEX CONCURRENTLY.
• В data migration не импортируйте модель напрямую — берите через apps.get_model (историческая версия модели).

⚠️ Ловушка: добавление столбца с НЕ-константным дефолтом или индекса без CONCURRENTLY берёт долгую блокировку и кладёт сервис. Большие data migrations лучше гонять батчами вне основной миграции.

Короткий ответ: Пул соединений переиспользует уже открытые соединения с БД, чтобы не платить за установку нового на каждый запрос. В SQLAlchemy пулом владеет Engine (QueuePool по умолчанию). В Django пул настраивается через CONN_MAX_AGE (persistent connections) или внешний пулер (PgBouncer).

Подробно:

# SQLAlchemy
engine = create_engine(
    url,
    pool_size=5,        # постоянных соединений
    max_overflow=10,    # сверх pool_size при пике
    pool_timeout=30,    # ждать соединение, сек
    pool_recycle=1800,  # пересоздать соединение через N сек (защита от обрыва)
    pool_pre_ping=True, # проверять «живость» перед выдачей
)

# Django settings.py
DATABASES = {"default": {..., "CONN_MAX_AGE": 60}}  # держать соединение 60 сек

⚠️ Ловушка: в serverless/многопроцессных средах (Lambda, много воркеров Gunicorn) число коннектов = воркеры × pool_size может превысить лимит БД (max_connections). Там часто ставят внешний пулер (PgBouncer) и маленький пул на процесс. Также пул нельзя шарить между процессами после fork — пересоздавайте Engine в воркере.

Короткий ответ: Сырые запросы выполняют через text()/execute() (SQLAlchemy) и raw()/cursor.execute() (Django). Главное правило — передавать данные ТОЛЬКО параметрами, никогда не подставлять f-string в SQL.

Подробно:

# ОПАСНО — SQL-инъекция
session.execute(text(f"SELECT * FROM users WHERE name = '{name}'"))  # НИКОГДА

# БЕЗОПАСНО — параметры
session.execute(text("SELECT * FROM users WHERE name = :name"), {"name": name})

# Django — безопасно
User.objects.raw("SELECT * FROM users WHERE name = %s", [name])

with connection.cursor() as cur:
    cur.execute("SELECT * FROM users WHERE name = %s", [name])

Если нужно подставить имя таблицы/колонки (их параметром не передать) — используйте белый список идентификаторов или psycopg.sql.Identifier, а не конкатенацию.

⚠️ Ловушка: %s в DB-API — это плейсхолдер параметра, а НЕ Python %-форматирование. cur.execute("... %s" % value) = инъекция; правильно cur.execute("... %s", [value]).

Короткий ответ: Если нужно лишь узнать «есть ли хоть одна строка» — используйте exists() (SELECT 1 ... LIMIT 1), а не count() (полный подсчёт) и не len(list(qs)) (загрузка всех строк). count() дорогой на больших таблицах.

Подробно:

# Django
if User.objects.filter(active=True).exists():   # SELECT 1 ... LIMIT 1 — дёшево
    ...
n = User.objects.filter(active=True).count()     # SELECT COUNT(*) — дороже
bad = len(User.objects.filter(active=True))      # грузит ВСЕ строки в память — худший вариант

# SQLAlchemy
from sqlalchemy import func, select, exists
has = session.scalar(select(exists().where(User.active == True)))
n = session.scalar(select(func.count()).select_from(User))

count() дорогой, потому что в PostgreSQL COUNT(*) обычно сканирует таблицу/индекс (нет дешёвого точного счётчика из-за MVCC). Для приблизительного числа используют статистику (pg_class.reltuples).

⚠️ Ловушка: if qs.count() > 0 для проверки наличия — антипаттерн: считает все строки, когда хватило бы exists(). И count() сразу после итерации того же qs не использует кэш — это отдельный запрос.

Короткий ответ: Soft delete — не удалять строку физически, а помечать её флагом (is_deleted/deleted_at). Доступ к «живым» строкам — через дефолтный фильтр (custom manager в Django, query-фильтр/событие в SQLAlchemy).

Подробно:

# Django
class SoftDeleteManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(deleted_at__isnull=True)

class Article(models.Model):
    deleted_at = models.DateTimeField(null=True, blank=True)
    objects = SoftDeleteManager()       # только живые
    all_objects = models.Manager()      # все, включая удалённые

    def delete(self, *a, **kw):
        self.deleted_at = timezone.now()
        self.save()

# SQLAlchemy — фильтрация при запросе
session.scalars(select(Article).where(Article.deleted_at.is_(None)))
# можно автоматизировать через with_loader_criteria / events

⚠️ Ловушка: UNIQUE-ограничения и soft delete конфликтуют — «удалённая» строка с тем же email мешает создать новую. Решение: partial unique index (WHERE deleted_at IS NULL). Также легко случайно показать удалённые данные, если забыть фильтр в сыром запросе или join.

Короткий ответ: Оптимистичная блокировка — вместо блокировки строки добавляют колонку версии; при UPDATE проверяют, что версия не изменилась (WHERE id=? AND version=?). Если 0 строк обновлено — кто-то успел раньше, бросаем ошибку конкуренции.

Подробно:

# SQLAlchemy — встроенная поддержка
class Account(Base):
    __tablename__ = "accounts"
    id = mapped_column(Integer, primary_key=True)
    balance = mapped_column(Integer)
    version_id = mapped_column(Integer, nullable=False)
    __mapper_args__ = {"version_id_col": version_id}
# UPDATE ... SET balance=?, version_id=version_id+1 WHERE id=? AND version_id=?
# если строк затронуто 0 -> StaleDataError

# Django — вручную через update с проверкой версии
updated = Account.objects.filter(id=acc.id, version=acc.version).update(
    balance=new_balance, version=F("version") + 1
)
if updated == 0:
    raise ConcurrencyError("Объект изменён другим процессом")

Optimistic подходит, когда конфликты редки (нет блокировок, лучше throughput). Альтернатива — pessimistic locking (SELECT ... FOR UPDATE / select_for_update()), когда конфликты часты.

⚠️ Ловушка: при оптимистичной блокировке нужно обрабатывать конфликт — обычно retry бизнес-операции. Без обработки ошибки пользователь просто получит исключение вместо корректного повтора.

Короткий ответ: one-to-many — FK на «многой» стороне; one-to-one — FK с unique; many-to-many — отдельная связующая таблица. Для M2M с доп.полями используют явную through/association-таблицу.

Подробно:

# Django
class Author(models.Model): ...
class Book(models.Model):
    author = models.ForeignKey(Author, on_delete=models.CASCADE)   # one-to-many

class Profile(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE)    # one-to-one

class Student(models.Model):
    courses = models.ManyToManyField("Course", through="Enrollment")  # M2M с полями

class Enrollment(models.Model):       # through-таблица с доп. данными
    student = models.ForeignKey(Student, on_delete=models.CASCADE)
    course = models.ForeignKey(Course, on_delete=models.CASCADE)
    grade = models.CharField(max_length=2)

# SQLAlchemy — many-to-many через association table
association = Table("student_course", Base.metadata,
    Column("student_id", ForeignKey("students.id"), primary_key=True),
    Column("course_id", ForeignKey("courses.id"), primary_key=True),
)
class Student(Base):
    courses = relationship("Course", secondary=association, back_populates="students")
# если у связи есть свои поля (grade) — association object вместо secondary

⚠️ Ловушка: простой ManyToManyField нельзя расширить полями постфактум — нужен through. И on_delete у FK — обязателен и определяет поведение (CASCADE/PROTECT/SET_NULL); неверный выбор приводит либо к потере данных, либо к ошибкам удаления.

Короткий ответ: Плохой SQL появляется при N+1, при размножении строк в JOIN-ах, при ненужном SELECT *, при сортировке/GROUP BY без индексов, при count()/len() там, где хватило бы exists(). Увидеть — через логирование SQL: echo=True в SQLAlchemy, django-debug-toolbar, логгер django.db.backends.

Подробно:

# SQLAlchemy — вывести весь SQL
engine = create_engine(url, echo=True)
# или логгер 'sqlalchemy.engine'

# Django — увидеть SQL запроса
print(qs.query)                       # сгенерированный SQL
from django.db import connection
print(connection.queries)             # все запросы за время (при DEBUG=True)
# django-debug-toolbar показывает кол-во запросов, дубли, время

Дальше — EXPLAIN:

print(qs.explain())                   # Django: план запроса

Симптомы плохого SQL: сотни одинаковых запросов (N+1), один запрос на много секунд (нет индекса/seq scan), огромный результат из-за JOIN коллекций.

⚠️ Ловушка: connection.queries наполняется только при DEBUG=True и копит запросы в памяти — на проде это и неинформативно, и опасно (рост памяти). Для прод-наблюдаемости используйте APM/slow query log БД.

Короткий ответ: Загруженный объект — это снимок строки на момент чтения. Если строку поменял другой процесс, ваш объект становится stale (устаревшим). ORM не обновляет его автоматически — нужно refresh/expire или повторный запрос.

Подробно:

# SQLAlchemy
acc = session.get(Account, 1)     # balance=100 в памяти
# ... другой процесс сделал balance=50 и закоммитил ...
print(acc.balance)                # всё ещё 100 (stale) пока объект не expired
session.refresh(acc)              # перечитать из БД -> 50
session.expire(acc)               # пометить устаревшим, перечитается при обращении

# Django
obj.refresh_from_db()             # перечитать из БД

Причины stale data:

• Долго живущий объект/сессия.
• Кэш на уровне приложения.
• Read replica с задержкой репликации (replication lag).

⚠️ Ловушка: read-modify-write на stale-объекте = lost update. Пример: прочитали balance, вычли в Python, сохранили — потеряли чужое изменение. Лечится F()-выражениями (вычисление в БД), SELECT FOR UPDATE или optimistic locking.

Короткий ответ: ORM делает дорогие операции синтаксически дешёвыми: навигация по связи = скрытый запрос (N+1), for obj in Model.objects.all() = загрузка всей таблицы в память, ленивые поля = повторные запросы. Удобство маскирует стоимость.

Подробно: Главные источники проблем:

• N+1 — самый частый, из-за ленивой навигации.
• Загрузка лишних данных — SELECT * вместо нужных колонок; гигантские результаты в память.
• Размножение строк в JOIN-ах при eager-загрузке коллекций.
• Лишние объекты — материализация моделей там, где хватило бы values()/агрегата в БД.
• count()/len() вместо exists().
• Запросы в цикле вместо одного запроса с агрегацией/bulk-операцией.

Противоядия: профилировать SQL, eager loading осознанно, агрегировать в БД (annotate/aggregate, func.*), bulk-операции, индексы, only/defer/values, для горячих/сложных мест — сырой SQL.

⚠️ Ловушка: «оптимизировать заранее» тоже вредно — eager loading всего подряд и преждевременные индексы. Сначала измерьте (debug-toolbar/echo/EXPLAIN), потом точечно чините узкие места.