Как минимизировать количество запросов в связанной таблице?

Как минимизировать количество запросов в связанной таблице

Проблема многократных запросов к БД при работе с related объектами — одна из самых частых причин медлительности приложений. Давайте разберём стратегии оптимизации.

Проблема: N+1 Query Problem

Основная проблема возникает, когда ты загружаешь список объектов, а потом для каждого загружаешь related данные:

# Плохо: N+1 queries
users = User.query.all()  # 1 запрос
for user in users:  # N запросов в цикле
    print(user.posts)  # Lazy loading — запрос за каждым пользователем!

# Total: 1 + N запросов

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

SQLAlchemy с явным join:

# Хорошо: 1-2 запроса
from sqlalchemy.orm import joinedload

users = session.query(User).options(joinedload(User.posts)).all()
for user in users:  # Данные уже загружены!
    print(user.posts)  # Нет дополнительных запросов

Этот подход использует LEFT JOIN и загружает все данные в один или два запроса.

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

Когда нужно фильтровать по related таблице:

from sqlalchemy.orm import contains_eager

# Загрузить пользователей, у которых есть посты с определённым статусом
users = session.query(User).join(
    User.posts
).filter(
    Post.status == 'published'
).options(
    contains_eager(User.posts)
).all()

Решение 3: Batch Loading (selectinload)

Это более гибкий подход — вторая query загружает батчами:

from sqlalchemy.orm import selectinload

users = session.query(User).options(
    selectinload(User.posts)
).all()

# Генерирует:
# SELECT * FROM users
# SELECT * FROM posts WHERE user_id IN (...)

Солидный выбор, когда нужна фильтрация по основной таблице.

Решение 4: Lazy Loading с select_in

Для более селективной загрузки:

from sqlalchemy import select
from sqlalchemy.orm import selectin_polymorphic

stmt = select(User).options(
    selectinload(User.posts.and_(Post.status == 'published'))
)
users = session.execute(stmt).scalars().unique().all()

Решение 5: Raw SQL для максимальной производительности

Когда нужна максимальная оптимизация:

# Один query с aggregation
users_with_posts = session.execute(
    """
    SELECT u.id, u.name, COUNT(p.id) as post_count
    FROM users u
    LEFT JOIN posts p ON u.id = p.user_id
    GROUP BY u.id
    """
).fetchall()

Решение 6: Caching (Redis/Memcached)

Для часто используемых данных:

import functools
from typing import List

def cache_result(ttl=3600):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Cache key = function name + args
            cache_key = f"{func.__name__}:{args}:{kwargs}"
            
            # Проверяем Redis
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
            
            # Если нет — выполняем запрос
            result = func(*args, **kwargs)
            redis_client.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@cache_result(ttl=3600)
def get_user_with_posts(user_id: int):
    return session.query(User).options(
        selectinload(User.posts)
    ).filter(User.id == user_id).first()

Решение 7: Pagination для больших наборов

Не загружай всё сразу:

from sqlalchemy import func

page = 1
page_size = 20

# Загружаем только нужную страницу
users = session.query(User).options(
    selectinload(User.posts)
).paginate(page=page, per_page=page_size)

# Total query:
# SELECT * FROM users LIMIT 20 OFFSET 0
# SELECT * FROM posts WHERE user_id IN (...)

Решение 8: GraphQL N+1 Solution

Если используешь GraphQL (например, Strawberry):

from strawberry import field
from strawberry.dataloader import DataLoader

class User:
    id: int
    name: str
    
    @field
    async def posts(self, info) -> List['Post']:
        # DataLoader автоматически батчит запросы
        loader = info.context.get_loader('posts')
        return await loader.load(self.id)

Comparison: Когда что использовать

# joinedload — когда нужны ВСЕ related данные
users = session.query(User).options(joinedload(User.posts)).all()

# selectinload — стандартный выбор (безопаснее для больших отношений)
users = session.query(User).options(selectinload(User.posts)).all()

# contains_eager — когда фильтруешь по related данным
users = session.query(User).join(User.posts).options(
    contains_eager(User.posts)
).filter(Post.status == 'published').all()

# Raw SQL — когда нужна максимальная оптимизация
# SELECT u.*, p.* FROM users u LEFT JOIN posts p ON u.id = p.user_id

# Caching — для часто используемых данных
@cache_result(ttl=3600)
def get_popular_posts():
    ...

Best Practices

• Профилируй запросы — используй sqlalchemy.event.listen() для логирования
• Индексы на внешних ключах — обязательно
• Batch size — выбери оптимальный размер батча
• Денормализация — иногда избыточна, но может помочь
• Connection pooling — правильная настройка пула соединений

# Логирование всех запросов
import logging

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

# Теперь видишь все запросы
users = session.query(User).options(selectinload(User.posts)).all()