Как решается проблема N+1 запросов в SQLAlchemy?

Как решается проблема N+1 запросов в SQLAlchemy

Что такое проблема N+1

# Вот классический N+1:
users = session.query(User).all()  # 1 запрос

for user in users:  # 100 пользователей
    print(user.posts)  # КАЖДЫЙ раз делается отдельный запрос!
    # Итого: 1 + 100 = 101 запрос!

Вместо одного запроса, который бы загрузил всё вместе, делаем 101 запрос.

Решение 1: Eager Loading с joinedload

from sqlalchemy.orm import joinedload

# Загружаем пользователей и их посты в один запрос
users = session.query(User).options(joinedload(User.posts)).all()

for user in users:
    print(user.posts)  # Уже в памяти, нет новых запросов

Подробнее:

from sqlalchemy.orm import joinedload

# Простая связь (один ко многим)
users = session.query(User).options(
    joinedload(User.posts)
).all()

# Вложенные связи
users = session.query(User).options(
    joinedload(User.posts).joinedload(Post.comments)
).all()

# Несколько связей
users = session.query(User).options(
    joinedload(User.posts),
    joinedload(User.profile)
).all()

Решение 2: Contains eager (для фильтрации)

Когда нужно загрузить только определённые связанные объекты:

from sqlalchemy.orm import contains_eager

# Загружаем только пользователей с постами за последний месяц
from datetime import datetime, timedelta

last_month = datetime.utcnow() - timedelta(days=30)

users = session.query(User).join(Post).filter(
    Post.created_at >= last_month
).options(
    contains_eager(User.posts)
).all()

Решение 3: Selectinload (для разных случаев)

Лучше для many-to-many и когда joinedload не подходит:

from sqlalchemy.orm import selectinload

# Два запроса вместо N+1:
# 1. SELECT * FROM users
# 2. SELECT posts.* FROM posts WHERE posts.user_id IN (...)
users = session.query(User).options(
    selectinload(User.posts)
).all()

for user in users:
    print(user.posts)  # Нет новых запросов

Решение 4: Raiseload (для отладки)

Вызывает исключение если попытаться загрузить ленивую связь:

from sqlalchemy.orm import raiseload

users = session.query(User).options(
    raiseload(User.posts)  # Выбросит ошибку если обратиться к user.posts
).all()

# Это помогает найти N+1 в коде

Решение 5: Явная загрузка через join

from sqlalchemy.orm import joinedload

# Загружаем пользователей и считаем их посты
from sqlalchemy import func

result = session.query(
    User,
    func.count(Post.id).label('post_count')
).outerjoin(Post).group_by(User.id).all()

for user, post_count in result:
    print(f"{user.name}: {post_count} posts")

Решение 6:批量 загрузка (batch loading)

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

# Вместо N+1 - делаем 2 запроса
users = session.query(User).all()

# Запрос 1: получить всех пользователей
user_ids = [u.id for u in users]

# Запрос 2: загрузить все посты для этих пользователей
posts = session.query(Post).filter(Post.user_id.in_(user_ids)).all()

# Организуем в словарь
posts_by_user = {}
for post in posts:
    if post.user_id not in posts_by_user:
        posts_by_user[post.user_id] = []
    posts_by_user[post.user_id].append(post)

# Теперь можем использовать
for user in users:
    user_posts = posts_by_user.get(user.id, [])

Практический пример

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, Session, joinedload
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="posts")

# ПЛОХО: N+1 запросов
def get_users_bad(session: Session):
    users = session.query(User).all()  # 1 запрос
    for user in users:
        print(user.posts)  # N запросов
    return users

# ХОРОШО: только 1 запрос
def get_users_good(session: Session):
    users = session.query(User).options(
        joinedload(User.posts)
    ).all()
    for user in users:
        print(user.posts)  # Нет новых запросов
    return users

# ХОРОШО: 2 запроса
def get_users_selectin(session: Session):
    users = session.query(User).options(
        selectinload(User.posts)
    ).all()
    for user in users:
        print(user.posts)
    return users

Когда какое решение использовать

joinedload:

• Один-ко-многим связи
• Когда нужно всё в одном запросе (LEFT OUTER JOIN)
• Может дублировать данные родителя

selectinload:

• Many-to-many связи
• Когда нужны отдельные запросы
• Когда joinedload может быть неэффективен

contains_eager:

• Когда нужна фильтрация связанных объектов
• Когда используешь join в where clause

Отладка

# Включи логирование SQL
import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# Или используй echo
session = Session(engine, echo=True)

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

1. Всегда думай о связях - не ленись их загружать
2. Профилируй - смотри сколько запросов на самом деле
3. Используй raiseload в development - ловит N+1
4. Кэшируй - если возможно
5. Пересмотри архитектуру - может быть иной способ

# Пример: лучше денормализовать
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    post_count = Column(Integer, default=0)  # Кэшируем

Это один из самых важных вопросов в ORM!