Оглавление

• Зачем бизнесу это нужно
• Что такое N+1 и как это выглядит на практике
• Как находить N+1: логи, профилирование, тесты
  • Пример: счётчик SQL в SQLAlchemy (самодостаточный пример)
• Django ORM: как убрать N+1 и не сломать логику
  • Типичная проблема N+1
  • Правильно: select_related и prefetch_related
  • Частые ловушки
• SQLAlchemy: joinedload/selectinload и когда что применять
• Массовые операции вместо циклов
• Кэширование, которое работает: уровни, инвалидация, защита от «набега»
  • Redis, простая обёртка и защита от «набега на кэш»
  • Выбор TTL и что кэшировать
• Безопасность и консистентность кэша
• Автотесты, которые ловят регрессии
  • Django: assertNumQueries
  • SQLAlchemy: собственный контекст подсчёта
• Чеклист внедрения
• Экономика: во сколько обходится N+1 и как быстро окупается работа
• Выводы

Зачем бизнесу это нужно

• Быстрее API — выше конверсия и меньше отток. Пользователь ждёт миллисекунды, не секунды.
• Меньше запросов к базе — меньше счет за облако и меньше рисков упереться в лимиты.
• Простые правки в ORM + кэш дают эффект за дни, а не за месяцы и без больших закупок железа.

Что такое N+1 и как это выглядит на практике

N+1 — это когда вы сначала делаете один запрос за списком объектов, а потом для каждого объекта ещё один (или несколько). Например, список статей и для каждой — автор и теги. В итоге вместо 1–2 запросов получаете десятки или сотни.

Симптомы:

• Страница со списком записей медленная именно при большом количестве элементов.
• В логах десятки одинаковых запросов «по одному и тому же шаблону».
• Растут задержки в пиковые часы из‑за нагрузки на базу.

Как находить N+1: логи, профилирование, тесты

• Включите логирование SQL. Для Django это LOGGING + запись django.db.backends. Для SQLAlchemy — echo=True или обработчик событий.
• Поставьте «счётчик запросов» в тесты и падайте, если их стало больше (регрессия).
• Прогоните страницу через локальный профайлер и посмотрите, что происходит с запросами при увеличении размера выборки (10 → 100).

Пример: счётчик SQL в SQLAlchemy (самодостаточный пример)

# python -m pip install sqlalchemy
from contextlib import contextmanager
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, Session, joinedload
from sqlalchemy import event

Base = declarative_base()

class Author(Base):
    __tablename__ = "author"
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)

class Article(Base):
    __tablename__ = "article"
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey("author.id"), nullable=False)
    author = relationship("Author", backref="articles")

engine = create_engine("sqlite:///:memory:", echo=False, future=True)
Base.metadata.create_all(engine)

# Наполним данными
with Session(engine) as s:
    authors = [Author(name=f"Author {i}") for i in range(10)]
    s.add_all(authors)
    s.flush()
    arts = [Article(title=f"Post {i}", author_id=authors[i % 10].id) for i in range(100)]
    s.add_all(arts)
    s.commit()

@contextmanager
def count_sql(engine):
    counter = {"n": 0}
    def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
        counter["n"] += 1
    h = event.listen(engine, "before_cursor_execute", before_cursor_execute)
    try:
        yield counter
    finally:
        event.remove(engine, "before_cursor_execute", before_cursor_execute)

# N+1
with Session(engine) as s, count_sql(engine) as c:
    articles = s.query(Article).all()  # 1 запрос
    authors = [a.author.name for a in articles]  # +100 запросов
    print("N+1 queries:", c["n"])  # ~101

# Оптимизировано: joinedload
from sqlalchemy.orm import selectinload
with Session(engine) as s, count_sql(engine) as c:
    articles = s.query(Article).options(joinedload(Article.author)).all()  # 1-2 запроса
    authors = [a.author.name for a in articles]
    print("Optimized queries:", c["n"])  # ~1

Django ORM: как убрать N+1 и не сломать логику

Представим простые модели.

# models.py
from django.db import models

class Author(models.Model):
    name = models.CharField(max_length=200)

class Tag(models.Model):
    name = models.CharField(max_length=50, unique=True)

class Article(models.Model):
    title = models.CharField(max_length=200)
    author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name="articles")
    tags = models.ManyToManyField(Tag, related_name="articles")
    published_at = models.DateTimeField()

Типичная проблема N+1

# views.py (антипаттерн)
from .models import Article

def list_titles_and_meta():
    result = []
    for a in Article.objects.order_by('-published_at')[:100]:
        result.append({
            "id": a.id,
            "title": a.title,
            "author": a.author.name,  # +1 на каждый объект
            "tags": [t.name for t in a.tags.all()],  # ещё +1 на каждый объект
        })
    return result

Получаем 1 запрос за список + до 200 дополнительных.

Правильно: select_related и prefetch_related

# views.py (оптимально)
from django.db.models import Prefetch
from .models import Article, Tag

def list_titles_and_meta_fast():
    qs = (
        Article.objects
        .select_related('author')
        .prefetch_related(
            Prefetch('tags', queryset=Tag.objects.only('id', 'name').order_by('name'))
        )
        .only('id', 'title', 'author__name', 'published_at')
        .order_by('-published_at')
    )
    result = []
    for a in qs[:100]:
        result.append({
            "id": a.id,
            "title": a.title,
            "author": a.author.name,
            "tags": [t.name for t in a.tags.all()],
        })
    return result

Замечания:

• select_related подходит для ForeignKey/OneToOne (подтягивает JOIN‑ом).
• prefetch_related — для ManyToMany/обратных связей (делает отдельные запросы, но один раз, с «раскладкой» в память).
• only/defer сокращают набор столбцов.

Частые ловушки

• Вы сортируете по полю, которого нет в only — Django сделает дополнительный запрос. Включайте поле сортировки в only.
• Счётчики вида obj.tags.count() в цикле — это тоже N+1. Либо аннотация с Count, либо prefetch и len(a.tags.all()).

SQLAlchemy: joinedload/selectinload и когда что применять

• joinedload — делает JOIN, нужен для «один-к-одному» и «многие-к-одному», когда связей немного.
• selectinload — отдельный запрос с IN по ключам, часто лучше для «один‑ко‑многим» и «многие‑ко‑многим», особенно если связей много.
• subqueryload — как selectinload, но через подзапрос; полезен реже.

Пример:

# Предположим модели Article/Author/Tag с отношениями
from sqlalchemy.orm import Session, joinedload, selectinload

# Список статей с автором и тегами, без N+1
with Session(engine) as s:
    articles = (
        s.query(Article)
        .options(
            joinedload(Article.author),
            selectinload(Article.tags)
        )
        .order_by(Article.id.desc())
        .limit(100)
        .all()
    )
    data = [{
        "id": a.id,
        "title": a.title,
        "author": a.author.name,
        "tags": [t.name for t in a.tags]
    } for a in articles]

Совет: если результат «раздувается» из‑за JOIN, переключайтесь на selectinload.

Массовые операции вместо циклов

• Вместо цикла с update по одному объекту — один update с фильтром.
• Вместо поштучного получения сущностей — один запрос с WHERE id IN (...).

Пример Django:

# увеличить счётчик просмотров у множества статей
from django.db.models import F
from .models import Article

Article.objects.filter(id__in=[1, 2, 3]).update(views_count=F('views_count') + 1)

Пример SQLAlchemy:

from sqlalchemy.orm import Session
from sqlalchemy import update

with Session(engine) as s:
    stmt = update(Article).where(Article.id.in_([1, 2, 3])).values(views_count=Article.views_count + 1)
    s.execute(stmt)
    s.commit()

Кэширование, которое работает: уровни, инвалидация, защита от «набега»

Кэш не лечит плохие запросы, но закрепляет результат оптимизации:

• Кэш объекта: быстрые карточки по id.
• Кэш списков: результаты популярных выборок (например, «последние 100»).
• Кэш «отрицательных» результатов: нет объекта — не ходим в БД каждый раз.

Redis, простая обёртка и защита от «набега на кэш»

# python -m pip install redis
# simple_cache.py
import json
import time
import redis

r = redis.Redis(host="localhost", port=6379, db=0)

class Cache:
    def __init__(self, client: redis.Redis):
        self.r = client

    def get_json(self, key: str):
        val = self.r.get(key)
        return json.loads(val) if val else None

    def set_json(self, key: str, value, ttl: int):
        self.r.setex(key, ttl, json.dumps(value, ensure_ascii=False))

    def with_lock(self, lock_key: str, timeout: int = 5):
        return self.r.lock(lock_key, timeout=timeout)

cache = Cache(r)

Кэш объекта в Django с инвалидацией по событию

# cache_article.py
from django.db.models.signals import post_save, post_delete, m2m_changed
from django.dispatch import receiver
from .models import Article
from .simple_cache import cache

ARTICLE_TTL = 300  # 5 минут

def article_cache_key(article_id: int) -> str:
    return f"article:{article_id}:v1"

def get_article_cached(article_id: int):
    key = article_cache_key(article_id)
    data = cache.get_json(key)
    if data is not None:
        return data

    lock_key = f"lock:{key}"
    with cache.with_lock(lock_key, timeout=5):
        # двойная проверка (другой поток мог уже положить)
        data = cache.get_json(key)
        if data is not None:
            return data
        # грузим из БД одной пачкой, без N+1
        from django.db.models import Prefetch
        from .models import Tag
        a = (Article.objects
             .select_related('author')
             .prefetch_related(Prefetch('tags', queryset=Tag.objects.only('id', 'name')))
             .only('id', 'title', 'author__name', 'published_at')
             .get(id=article_id))
        data = {
            "id": a.id,
            "title": a.title,
            "author": a.author.name,
            "published_at": a.published_at.isoformat(),
            "tags": [t.name for t in a.tags.all()],
        }
        cache.set_json(key, data, ARTICLE_TTL)
        return data

# Инвалидация при изменениях
@receiver(post_save, sender=Article)
@receiver(post_delete, sender=Article)
def invalidate_article_cache(sender, instance, **kwargs):
    cache.r.delete(article_cache_key(instance.id))

# Если меняются теги статьи — тоже инвалидируем
@receiver(m2m_changed, sender=Article.tags.through)
def invalidate_article_tags_cache(sender, instance, action, **kwargs):
    if action in {"post_add", "post_remove", "post_clear"}:
        cache.r.delete(article_cache_key(instance.id))

Кэш списков с версионным ключом (без опасных шаблонных удалений)

# list_cache.py
from .simple_cache import cache
from .models import Article

LIST_TTL = 120
LIST_VERSION_KEY = "articles:list:v"

def _get_list_version() -> int:
    v = cache.r.get(LIST_VERSION_KEY)
    if v is None:
        cache.r.set(LIST_VERSION_KEY, 1)
        return 1
    return int(v)

def _bump_list_version():
    cache.r.incr(LIST_VERSION_KEY)

def list_cache_key(suffix: str) -> str:
    return f"articles:list:{_get_list_version()}:{suffix}"

def get_latest_articles_cached(limit: int = 100):
    key = list_cache_key(f"latest:{limit}")
    data = cache.get_json(key)
    if data is not None:
        return data
    # защищаемся от набега
    with cache.with_lock(f"lock:{key}", timeout=5):
        data = cache.get_json(key)
        if data is not None:
            return data
        qs = (Article.objects
              .select_related('author')
              .only('id', 'title', 'author__name', 'published_at')
              .order_by('-published_at')[:limit])
        data = [{
            "id": a.id,
            "title": a.title,
            "author": a.author.name,
            "published_at": a.published_at.isoformat(),
        } for a in qs]
        cache.set_json(key, data, LIST_TTL)
        return data

# вызывать при изменении каталога (сигналы post_save/post_delete Article или по расписанию)
def on_articles_changed():
    _bump_list_version()

Почему версионный ключ лучше, чем удалить «все по маске»:

• Не трогаем блокировку Redis на SCAN/KEYS, не ловим таймауты.
• Один INCR мгновенно «устарит» все старые списки.

Выбор TTL и что кэшировать

• Короткий TTL (1–5 минут) подходит для динамических списков.
• Долгий TTL (час/сутки) — для редко меняющихся карточек, но не забывайте про инвалидацию по событиям.
• Кэшируйте «нет данных» (например, 404) на 10–30 секунд — это снимает горячие запросы «в никуда».

Безопасность и консистентность кэша

• Не кладите в кэш персональные данные без шифрования и явного решения, где хранится ключ (KMS/конфиг). Лучше кэшировать безопасные проекции (имя автора, заголовок, дата).
• Разделяйте ключи по арендаторам/клиентам: tenant:123:article
  , чтобы исключить утечки между организациями.
• Храните только сериализованные данные (JSON), не складируйте объекты ORM.
• На запись — делайте «сначала БД, потом кэш». Инвалидация — delete/incr, не «сначала кэш, потом БД».

Автотесты, которые ловят регрессии

Django: assertNumQueries

# tests/test_queries.py
from django.test import TestCase
from django.db import connection
from app.models import Article
from app.views import list_titles_and_meta_fast

class QueryTests(TestCase):
    fixtures = ["demo.json"]  # Предположим, у вас есть тестовые данные

    def test_list_uses_few_queries(self):
        with self.assertNumQueries(3):
            data = list_titles_and_meta_fast()
            self.assertGreater(len(data), 0)

• Число 3 условно: 1 — выборка статей с автором, 1 — префетч тегов, 1 — мелкие служебные запросы (зависит от проекта). Подстройте под себя и закрепите лимит.

SQLAlchemy: собственный контекст подсчёта

В примере выше показан count_sql. Оборачивайте критичные запросы в тестах и фиксируйте верхнюю границу.

Чеклист внедрения

1. Включите лог SQL на тестовом окружении и соберите «топ медленных страниц/методов».
2. Для каждой — измерьте количество запросов на 10/50/100 элементов.
3. Примените select_related/prefetch_related (Django) или joinedload/selectinload (SQLAlchemy).
4. Замените циклы на массовые операции и IN‑запросы.
5. Добавьте кэш: карточки и популярные списки. Включите инвалидацию по событию и версионные ключи.
6. Защитите кэш от «набега»: локи/замки и повторная проверка перед записью.
7. Напишите тесты с лимитом на количество запросов.
8. Наблюдайте метрики: p95 латентности, RPS, нагрузка на базу (CPU/IO/активные коннекты), hit‑ratio кэша.

Экономика: во сколько обходится N+1 и как быстро окупается работа

• Было: 100 запросов к БД на один вызов API, 200 RPS → 20 000 запросов/с к БД в пике. Понадобится крупный инстанс базы и пул соединений на сотни слотов.
• Стало: 2–3 запроса, плюс кэш с попаданием 70%. В БД летит менее 200–300 запросов/с при тех же 200 RPS. Инстанс меньшего размера, запас прочности выше.
• Реальная экономия: уменьшение класса БД/кластера, меньше реплик, меньше аварий ночью. Время отклика — с 400–700 мс до 80–150 мс на холодном кэше и до 20–40 мс на тёплом.

Даже неделя работы команды часто окупается за 1–2 месяца счетов за облако.

Выводы

• N+1 — простой источник больших проблем. Его легко диагностировать и убрать силами ORM.
• Кэш закрепляет результат: карточки и списки с безопасной инвалидацией и защитой от «набега».
• Закройте тему автотестами и наблюдаемостью, и вы удержите SLA без докупки серверов.