Как избавиться от N плюс 1?

Проблема N+1 запросов в SQL

Проблема N+1 — это классический антипаттерн в работе с базами данных, когда при загрузке коллекции объектов выполняется 1 запрос для получения родителей, а затем по N отдельному запросу для каждого дочернего объекта. Результат: вместо оптимальных 2-3 запросов выполняется N+1 запрос.

Пример проблемы

Плохо: N+1 запросы

# 1 запрос: SELECT * FROM users
users = session.execute(select(User)).scalars().all()

# N запросов: SELECT * FROM posts WHERE user_id = ?
for user in users:
    posts = user.posts  # Каждый доступ — отдельный запрос!
    print(f"{user.name}: {len(posts)} posts")

# Итого: 1 + N запросов (например, 1 + 1000 = 1001 запрос)

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

selectinload() — отдельный SELECT IN запрос

from sqlalchemy.orm import selectinload
from sqlalchemy import select

# Загружаем посты заранее одним SELECT IN запросом
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).unique().scalars().all()

for user in users:
    # posts уже загружены в памяти
    print(f"{user.name}: {len(user.posts)} posts")

# Итого: 2 запроса (1 для users + 1 для posts всех users)

joinedload() — JOIN запрос

from sqlalchemy.orm import joinedload

# Загружаем посты через JOIN
stmt = select(User).options(joinedload(User.posts))
users = session.execute(stmt).unique().scalars().all()

for user in users:
    print(f"{user.name}: {len(user.posts)} posts")

# Итого: 1 запрос (JOIN users с posts)

contains_eager() — для фильтрации

from sqlalchemy.orm import contains_eager
from sqlalchemy import select, join

# Фильтруем посты и загружаем их через JOIN
stmt = (
    select(User)
    .join(User.posts)
    .where(Post.published == True)
    .options(contains_eager(User.posts))
)
users = session.execute(stmt).unique().scalars().all()

# Итого: 1 запрос

Решение 2: Явный JOIN запрос

from sqlalchemy import select, func

# Получаем users с count posts в одном запросе
stmt = (
    select(
        User,
        func.count(Post.id).label('post_count')
    )
    .join(Post, User.id == Post.user_id, isouter=True)
    .group_by(User.id)
)

results = session.execute(stmt).all()
for user, post_count in results:
    print(f"{user.name}: {post_count} posts")

Решение 3: Батчинг (Batch Loading)

from sqlalchemy.orm import subqueryload

# Загружаем посты батчами (по N за раз)
stmt = select(User).options(subqueryload(User.posts))
users = session.execute(stmt).scalars().all()

for user in users:
    print(f"{user.name}: {len(user.posts)} posts")

# Итого: 2 запроса, но более оптимально при наличии фильтров

Решение 4: Ручной SQL с GROUP_CONCAT

from sqlalchemy import text

results = session.execute(text("""
    SELECT 
        u.id, 
        u.name,
        STRING_AGG(p.title, ', ') as post_titles
    FROM users u
    LEFT JOIN posts p ON u.id = p.user_id
    GROUP BY u.id, u.name
""")).all()

for row in results:
    print(f"{row.name}: {row.post_titles}")

Решение 5: DataLoader паттерн (для GraphQL)

from dataclasses import dataclass
from typing import List
from collections import defaultdict

@dataclass
class DataLoader:
    def load_posts(self, user_ids: List[int]) -> dict:
        """Загружаем посты для всех юзеров одним запросом"""
        stmt = select(Post).where(Post.user_id.in_(user_ids))
        posts = session.execute(stmt).scalars().all()
        
        # Группируем по user_id
        posts_by_user = defaultdict(list)
        for post in posts:
            posts_by_user[post.user_id].append(post)
        
        return posts_by_user

# Использование
loader = DataLoader()
user_ids = [1, 2, 3, 4, 5]
posts_by_user = loader.load_posts(user_ids)

for user_id, posts in posts_by_user.items():
    print(f"User {user_id}: {len(posts)} posts")

# Итого: 1 запрос для всех постов

Решение 6: Денормализация и кэширование

from sqlalchemy.orm import relationship
from sqlalchemy import Column, Integer

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    posts_count = Column(Integer, default=0)  # Кэшируемое значение
    
    posts = relationship("Post")

# Теперь можно быстро получить count без JOIN
users = session.execute(select(User)).scalars().all()
for user in users:
    print(f"{user.name}: {user.posts_count} posts")  # Нет доп. запроса

Практический пример: Оптимизированный сервис

from sqlalchemy.orm import selectinload
from sqlalchemy import select

class UserService:
    def get_users_with_posts(self) -> List[User]:
        """Получить всех пользователей с их постами"""
        # Правильный способ: eager loading
        stmt = select(User).options(selectinload(User.posts))
        return session.execute(stmt).scalars().unique().all()
    
    def get_active_users_with_published_posts(self) -> List[User]:
        """Получить активных пользователей с опубликованными постами"""
        stmt = (
            select(User)
            .where(User.is_active == True)
            .options(
                selectinload(User.posts).where(Post.published == True)
            )
        )
        return session.execute(stmt).scalars().unique().all()

# Использование
service = UserService()
users = service.get_users_with_posts()
for user in users:
    print(f"{user.name}: {len(user.posts)} posts")

Чеклист по предотвращению N+1

• Используй selectinload() или joinedload() при загрузке связанных объектов
• Профилируй запросы с помощью SQLAlchemy логирования
• Избегай доступа к атрибутам дочерних объектов в циклах без eager loading
• Используй contains_eager() если нужна фильтрация дочерних объектов
• Батчируй запросы если невозможно использовать JOIN
• Кэшируй часто запрашиваемые связи (денормализация)

Инструменты для отладки

from sqlalchemy import event
from sqlalchemy.engine import Engine

@event.listens_for(Engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    print(f"SQL: {statement}")

# Теперь видны все запросы

Проблема N+1 критична для производительности. Правильное использование eager loading экономит сотни/тысячи запросов в боевом коде.