Что такое N+1 проблема в ORM и как её решить?

N+1 проблема в ORM и её решение

N+1 проблема — это производственная проблема производительности, когда при загрузке N объектов выполняется N+1 запросов к БД вместо одного или двух.

Как возникает N+1

Пример с SQLAlchemy:

from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Author(Base):
    __tablename__ = "authors"
    
    id = Column(Integer, primary_key=True)
    name = Column(String)
    books = relationship("Book")  # Отношение к книгам

class Book(Base):
    __tablename__ = "books"
    
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey("authors.id"))

engine = create_engine("sqlite:///:memory:")
Session = sessionmaker(bind=engine)
session = Session()

# ❌ N+1 проблема
authors = session.query(Author).all()  # Запрос 1: SELECT * FROM authors

for author in authors:
    print(author.name)
    # Для каждого автора запрос к его книгам
    # Запросы 2 до N+1: SELECT * FROM books WHERE author_id = ?
    for book in author.books:
        print(f"  - {book.title}")

# Если 100 авторов — выполнится 101 запрос!

Решение 1: Eager Loading (JOIN)

Загружай связанные данные сразу:

from sqlalchemy.orm import joinedload

# ✅ Правильно: один запрос с JOIN
authors = session.query(Author).options(
    joinedload(Author.books)
).all()

for author in authors:
    print(author.name)
    for book in author.books:
        print(f"  - {book.title}")

# Выполнится ОДИН запрос с LEFT OUTER JOIN

SQL результат:

SELECT authors.id, authors.name, books.id, books.title, books.author_id
FROM authors
LEFT OUTER JOIN books ON authors.id = books.author_id

Решение 2: selectinload (для коллекций)

Для больших коллекций лучше чем JOIN:

from sqlalchemy.orm import selectinload

# ✅ Правильно: selectinload
authors = session.query(Author).options(
    selectinload(Author.books)
).all()

# Выполнится 2 запроса:
# 1. SELECT * FROM authors
# 2. SELECT * FROM books WHERE author_id IN (id1, id2, ...)

Это лучше, чем JOIN, когда:

• Много-ко-многим отношения
• Книг очень много (JOIN будет тяжёлый)
• Нужна паджинация

Решение 3: subqueryload

Для некоторых сложных отношений:

from sqlalchemy.orm import subqueryload

authors = session.query(Author).options(
    subqueryload(Author.books)
).all()

# Выполнится через подзапрос

Решение 4: Явный JOIN в query

from sqlalchemy.orm import contains_eager
from sqlalchemy import func

# ✅ Правильно: JOIN с contains_eager
authors = session.query(Author).join(
    Book  # Явно джойним
).options(
    contains_eager(Author.books)
).all()

# Или более читаемо:
from sqlalchemy import select, outerjoin

stmt = (
    select(Author).join(Book).options(
        contains_eager(Author.books)
    )
)
authors = session.execute(stmt).unique().scalars().all()

Решение 5: Select с relationship

Модерный подход (SQLAlchemy 2.0+):

from sqlalchemy import select
from sqlalchemy.orm import selectinload

# ✅ Правильно: select с selectinload
stmt = select(Author).options(
    selectinload(Author.books)
)

authors = session.execute(stmt).scalars().unique().all()

Решение 6: Только нужные поля

Загружайте только необходимые колонки:

# ❌ Плохо: загружаются все поля
authors = session.query(Author).all()

# ✅ Хорошо: только нужные поля
authors = session.query(Author.id, Author.name).all()

# ✅ Ещё лучше: с отношением
result = session.query(
    Author.id,
    Author.name,
    Book.title
).join(Book).all()

for author_id, author_name, book_title in result:
    print(f"{author_name}: {book_title}")

Решение 7: Batch запросы вручную

Для максимального контроля:

def get_authors_with_books(author_ids: list[int]):
    # Запрос 1: авторы
    authors = session.query(Author).filter(
        Author.id.in_(author_ids)
    ).all()
    
    # Запрос 2: все книги за раз
    books = session.query(Book).filter(
        Book.author_id.in_(author_ids)
    ).all()
    
    # Собираем в памяти (быстро)
    books_by_author = {}
    for book in books:
        if book.author_id not in books_by_author:
            books_by_author[book.author_id] = []
        books_by_author[book.author_id].append(book)
    
    # Присваиваем книги авторам
    for author in authors:
        author.books = books_by_author.get(author.id, [])
    
    return authors

# 2 запроса вместо N+1
author_ids = [1, 2, 3]
authors = get_authors_with_books(author_ids)

Решение 8: Lazy loading с populate_existing

Для динамической загрузки:

from sqlalchemy.orm import noload

# ✅ Правильно: отключить lazy loading
authors = session.query(Author).options(
    noload(Author.books)  # Не загружать книги
).all()

# Загружать книги только когда нужно
for author in authors:
    # Явная загрузка только для этого автора
    session.refresh(author, ["books"])
    print(author.books)

Решение 9: Кэширование на уровне приложения

from functools import lru_cache

@lru_cache(maxsize=128)
def get_author_books(author_id: int):
    """Кэшируем запрос."""
    return session.query(Book).filter(
        Book.author_id == author_id
    ).all()

# Первый вызов — запрос в БД
books = get_author_books(1)

# Второй вызов — из кэша (нет запроса в БД)
books = get_author_books(1)

Решение 10: Batch processing

Для обработки больших наборов данных:

def process_authors(batch_size=100):
    """Обработка авторов батчами."""
    offset = 0
    
    while True:
        # Загружаем батч
        authors = session.query(Author).options(
            selectinload(Author.books)
        ).offset(offset).limit(batch_size).all()
        
        if not authors:
            break
        
        # Обрабатываем
        for author in authors:
            print(f"{author.name}: {len(author.books)} books")
        
        offset += batch_size

Диагностика N+1

SQL логирование

import logging

logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# Теперь все SQL запросы выводятся в лог

Подсчёт запросов в тестах

from sqlalchemy import event

query_count = 0

@event.listens_for(engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    global query_count
    query_count += 1

# Ваш код
authors = session.query(Author).all()
for author in authors:
    for book in author.books:
        pass

print(f"Executed {query_count} queries")  # Покажет 101 для 100 авторов

pytest-django с assertNumQueries

from django.test import TestCase
from django.test.utils import assertNumQueries

class AuthorTests(TestCase):
    def test_no_n_plus_one(self):
        # Ожидаем только 2 запроса
        with self.assertNumQueries(2):
            authors = Author.objects.prefetch_related('books')
            for author in authors:
                for book in author.books:
                    pass

Сравнение методов

Лучшие практики

• Всегда профилируйте SQL запросы в development
• Используйте selectinload по умолчанию для отношений
• Помните о каскадных отношениях (отношение отношение отношение)
• Кэшируйте часто используемые данные
• Пишите тесты на количество запросов
• Мониторьте в production (новрелик, datadog)
• Не бойтесь SQL — иногда raw SQL быстрее

# ✅ Итоговый пример
from sqlalchemy.orm import selectinload

authors = session.query(Author).options(
    selectinload(Author.books).selectinload(Book.reviews)
).all()

# Загружает авторов, их книги и рецензии на книги за 3 запроса

N+1 проблема — это одна из самых распространённых проблем производительности в работе с ORM.