Какие знаешь методы оптимизации ORM?

Методы оптимизации ORM

ORM (Object-Relational Mapping) может быть медленной, если её неправильно использовать. Рассмотрю основные методы оптимизации для повышения производительности.

1. N+1 Query Problem

Проблема: Запрос в цикле — один запрос для объекта + N запросов для связанных объектов.

from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import declarative_base, relationship, Session

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    posts = relationship('Post')  # Связь один-ко-многим

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))

# ПЛОХО: N+1 проблема
users = session.query(User).all()  # 1 запрос
for user in users:  # N+1 = N запросов по одному для каждого пользователя
    print(user.posts)  # SELECT * FROM posts WHERE user_id = ?

# ХОРОШО: Eager loading
from sqlalchemy.orm import joinedload

users = session.query(User).options(joinedload(User.posts)).all()  # 1-2 запроса
for user in users:
    print(user.posts)  # Данные уже загружены

2. Eager Loading Strategies

joinedload — LEFT OUTER JOIN:

users = session.query(User).options(joinedload(User.posts)).all()
# SELECT users.id, users.name, posts.id, posts.title
# FROM users LEFT OUTER JOIN posts ON users.id = posts.user_id

selectinload — Отдельный SELECT с IN:

users = session.query(User).options(selectinload(User.posts)).all()
# Запрос 1: SELECT * FROM users
# Запрос 2: SELECT * FROM posts WHERE user_id IN (1, 2, 3, ...)

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

from sqlalchemy.orm import contains_eager

# Получить пользователей с постами из 2024 года
users = session.query(User).join(User.posts).filter(
    Post.created_at.year == 2024
).options(contains_eager(User.posts)).all()

3. Batch Loading

Проблема: Большой batch на памяти.

# ПЛОХО: всё в памяти
users = session.query(User).all()  # 1M пользователей — OOM
for user in users:
    process(user)

# ХОРОШО: batch processing
for batch_users in session.query(User).yield_per(1000):
    for user in batch_users:
        process(user)

4. Использование Raw SQL для сложных запросов

# Когда ORM становится слишком сложной
from sqlalchemy import text

# СЛОЖНО с ORM
result = session.query(User).filter(
    User.posts.any(
        Post.comments.any(
            Comment.rating > 5
        )
    )
).all()

# ПРОСТО с SQL
query = text("""
    SELECT DISTINCT u.*
    FROM users u
    JOIN posts p ON u.id = p.user_id
    JOIN comments c ON p.id = c.post_id
    WHERE c.rating > 5
""")
result = session.execute(query).fetchall()

5. Column Selection (Projection)

Проблема: Загружаем все колонки, даже если нужны только несколько.

# ПЛОХО: загружаются все колонки
users = session.query(User).all()
for user in users:
    print(user.name)  # Остальные поля не нужны

# ХОРОШО: загружаем только нужные колонки
names = session.query(User.id, User.name).all()
for id, name in names:
    print(name)

6. Индексы на БД

# Дефайнить индексы в модели
from sqlalchemy import Index

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    email = Column(String, index=True)  # Простой индекс
    status = Column(String)
    
    __table_args__ = (
        Index('idx_email_status', 'email', 'status'),  # Составной индекс
    )

# Проверить индексы
session.query(User).filter(User.email == 'test@example.com').explain()

7. Query Caching

from functools import lru_cache
import hashlib

# Простое кэширование результатов
class QueryCache:
    def __init__(self, session):
        self.session = session
        self.cache = {}
    
    def get_user(self, user_id):
        if user_id in self.cache:
            return self.cache[user_id]
        
        user = self.session.query(User).filter(User.id == user_id).first()
        self.cache[user_id] = user
        return user

# Или с Redis
import redis
import json

redis_client = redis.Redis()

def get_user(user_id):
    key = f"user:{user_id}"
    cached = redis_client.get(key)
    
    if cached:
        return json.loads(cached)
    
    user = session.query(User).filter(User.id == user_id).first()
    redis_client.setex(key, 3600, json.dumps(user.__dict__))
    return user

8. Lazy vs Eager Loading

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    
    # LAZY (по умолчанию) — загружается при обращении
    posts = relationship('Post', lazy='select')
    
    # EAGER — загружается сразу
    # posts = relationship('Post', lazy='joined')

9. Connection Pooling

from sqlalchemy.pool import QueuePool

engine = create_engine(
    'postgresql://user:password@localhost/db',
    poolclass=QueuePool,
    pool_size=10,  # Максимум активных коннекций
    max_overflow=20,  # Дополнительные коннекции если нужны
    pool_recycle=3600,  # Переиспользовать коннекцию каждый час
)

10. Batch Insert/Update

# ПЛОХО: медленно
for item in items:
    session.add(item)
    session.commit()  # Коммит после каждого

# ХОРОШО: batch
session.bulk_insert_mappings(Item, items)  # Не вызывает __init__
session.commit()

# Или с использованием execute
from sqlalchemy import insert

session.execute(
    insert(User),
    [{'name': 'user1'}, {'name': 'user2'}, ...]
)
session.commit()

11. Query Optimization

# Профилирование запросов
from sqlalchemy import event
from sqlalchemy.engine import Engine
import time

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())
    print(f"Query: {statement}")

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total_time = time.time() - conn.info['query_start_time'].pop(-1)
    print(f"Execution time: {total_time:.4f}s")

12. Pagination

# ПЛОХО: загружаем все записи
all_users = session.query(User).all()

# ХОРОШО: пагинация
page = 1
page_size = 20
users = session.query(User).offset((page-1)*page_size).limit(page_size).all()

# С общим количеством
total = session.query(User).count()
users = session.query(User).offset((page-1)*page_size).limit(page_size).all()

Чеклист оптимизации ORM

• Решить N+1 проблему с joinedload/selectinload
• Использовать batch processing для больших наборов
• Выбирать только нужные колонки (projection)
• Добавить индексы на часто фильтруемые поля
• Использовать raw SQL для сложных запросов
• Включить query logging для профилирования
• Настроить connection pooling
• Реализовать caching критических запросов
• Использовать batch insert/update вместо одиночных
• Добавить пагинацию для больших результатов

Таблица: Когда использовать что

Вывод: ORM — это удобство ценой производительности. Знайте ограничения и используйте raw SQL когда нужна скорость.