Что такое агрегация в ORM?

Что такое агрегация в ORM

Агрегация в ORM — это операция, которая объединяет и преобразует данные из одной или нескольких таблиц, используя функции типа COUNT, SUM, AVG, MIN, MAX. Это позволяет получить статистику и сводки без загрузки всех исходных записей в приложение.

Почему агрегация важна

Без агрегации нужно было бы:

1. Загрузить 1 миллион записей в память
2. Обработать их в Python
3. Вычислить результат

С агрегацией база данных сама вычисляет результат и отправляет нам только итоговое число — это намного быстрее и экономнее.

Основные функции агрегации

1. COUNT — подсчёт записей

from sqlalchemy import func, select
from sqlalchemy.orm import Session
from models import User

with Session(engine) as session:
    # Простой COUNT
    total_users = session.query(User).count()
    # или с новым синтаксисом SQLAlchemy 2.0
    total_users = session.execute(
        select(func.count(User.id))
    ).scalar()
    
    print(f"Всего пользователей: {total_users}")  # 15243
    
    # COUNT с условием
    active_users = session.execute(
        select(func.count(User.id)).where(User.is_active == True)
    ).scalar()
    
    print(f"Активных пользователей: {active_users}")  # 8921

2. SUM — сумма значений

from models import Order

# Сумма всех заказов
total_revenue = session.execute(
    select(func.sum(Order.amount))
).scalar() or 0  # or 0 на случай, если нет данных

print(f"Общая выручка: ${total_revenue}")

# SUM с фильтром
today_revenue = session.execute(
    select(func.sum(Order.amount)).where(
        func.date(Order.created_at) == datetime.now().date()
    )
).scalar() or 0

print(f"Выручка сегодня: ${today_revenue}")

3. AVG — среднее значение

# Средняя стоимость заказа
average_order = session.execute(
    select(func.avg(Order.amount))
).scalar()

print(f"Средний заказ: ${average_order:.2f}")

# Средний рейтинг продукта
average_rating = session.execute(
    select(func.avg(Review.rating)).where(Review.product_id == 42)
).scalar()

print(f"Средний рейтинг: {average_rating:.1f}/5")

4. MIN/MAX — минимум и максимум

from sqlalchemy import and_

# Минимальная и максимальная цена
min_price, max_price = session.execute(
    select(func.min(Product.price), func.max(Product.price))
).first()

print(f"Цены: от ${min_price} до ${max_price}")

# Самый новый заказ
latest_order = session.execute(
    select(func.max(Order.created_at))
).scalar()

print(f"Последний заказ: {latest_order}")

GROUP BY — группировка с агрегацией

Агрегация часто используется вместе с GROUP BY для получения статистики по категориям.

from sqlalchemy import func, select, desc
from models import Order, Product

# Сколько заказов по каждому дню?
orders_by_date = session.execute(
    select(
        func.date(Order.created_at).label("date"),
        func.count(Order.id).label("count")
    ).group_by(func.date(Order.created_at))
    .order_by(desc("date"))
)

for date, count in orders_by_date:
    print(f"{date}: {count} заказов")

# Выручка по продуктам
revenue_by_product = session.execute(
    select(
        Product.name,
        func.count(Order.id).label("orders_count"),
        func.sum(Order.amount).label("total_revenue"),
        func.avg(Order.amount).label("avg_order")
    ).join(Order, Order.product_id == Product.id)
    .group_by(Product.name)
    .order_by(desc("total_revenue"))
)

for name, count, revenue, avg in revenue_by_product:
    print(f"{name}: {count} заказов, ${revenue}, средний ${avg:.2f}")

HAVING — фильтрация по агрегированным значениям

HAVING похож на WHERE, но работает с результатами GROUP BY.

# Какие категории генерируют более $10,000 выручки?
from sqlalchemy import func, select, desc
from models import Order, Category

profitable_categories = session.execute(
    select(
        Category.name,
        func.sum(Order.amount).label("total_revenue")
    ).join(Order, Order.category_id == Category.id)
    .group_by(Category.id, Category.name)
    .having(func.sum(Order.amount) > 10000)  # Фильтруем по агрегированному значению
    .order_by(desc("total_revenue"))
)

for name, revenue in profitable_categories:
    print(f"{name}: ${revenue}")

# Какие пользователи сделали более 10 заказов?
from models import User

frequent_buyers = session.execute(
    select(
        User.email,
        func.count(Order.id).label("order_count")
    ).join(Order, Order.user_id == User.id)
    .group_by(User.id, User.email)
    .having(func.count(Order.id) > 10)
)

for email, count in frequent_buyers:
    print(f"{email}: {count} заказов")

Сложные агрегации с подзапросами

from sqlalchemy import select, func, and_
from models import Order, OrderItem, Product

# TOP-3 самые популярные товары по количеству проданных единиц
top_products = session.execute(
    select(
        Product.name,
        func.sum(OrderItem.quantity).label("total_sold")
    ).join(OrderItem, OrderItem.product_id == Product.id)
    .group_by(Product.id, Product.name)
    .order_by(func.sum(OrderItem.quantity).desc())
    .limit(3)
)

for name, sold in top_products:
    print(f"{name}: продано {sold} шт.")

# Средняя стоимость заказа по регионам
from models import User, Order

region_stats = session.execute(
    select(
        User.region,
        func.count(Order.id).label("order_count"),
        func.avg(Order.amount).label("avg_amount"),
        func.sum(Order.amount).label("total_amount")
    ).join(Order, Order.user_id == User.id)
    .group_by(User.region)
    .having(func.count(Order.id) >= 5)  # Только регионы с 5+ заказами
    .order_by(func.sum(Order.amount).desc())
)

for region, count, avg, total in region_stats:
    print(f"{region}: {count} заказов, среднее ${avg:.2f}, всего ${total}")

Сравнение: без агрегации vs с агрегацией

НЕПРАВИЛЬНО — загружаем всё в приложение

# Плохо: загружаем 1 миллион записей
orders = session.query(Order).all()  # МЕДЛЕННО!
total = sum(order.amount for order in orders)
count = len(orders)
average = total / count

print(f"Сумма: ${total}, Среднее: ${average}")

ПРАВИЛЬНО — агрегация на БД

# Хорошо: БД сама вычисляет результат
total, count, average = session.execute(
    select(
        func.sum(Order.amount),
        func.count(Order.id),
        func.avg(Order.amount)
    )
).first()

print(f"Сумма: ${total}, Среднее: ${average}")

Разница: первый вариант загружает 1 миллион записей в память (медленно), второй — одну строку с результатом (быстро).

Специальные функции агрегации

STRING_AGG — конкатенация строк

# Список всех email пользователей в одной строке
emails = session.execute(
    select(func.string_agg(User.email, ', '))
).scalar()

print(f"Письма: {emails}")  # user1@mail.com, user2@mail.com, ...

ARRAY_AGG — массив значений (PostgreSQL)

# Список всех заказов пользователя в виде массива
from sqlalchemy.dialects.postgresql import array_agg

user_orders = session.execute(
    select(
        User.email,
        func.array_agg(Order.id).label("order_ids")
    ).join(Order, Order.user_id == User.id)
    .group_by(User.id)
).first()

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

1. Всегда используй агрегацию вместо загрузки всех данных в Python
2. Добавляй индексы на колонки, по которым агрегируешь

CREATE INDEX idx_order_created_at ON orders(created_at);
CREATE INDEX idx_order_amount ON orders(amount);

1. Используй EXPLAIN для анализа запроса

from sqlalchemy import text

query = select(func.count(Order.id))
print(session.execute(text(f"EXPLAIN {query}")).fetchall())

1. Кешируй результаты агрегаций — они часто не меняются каждую секунду
2. Используй периодические отчёты вместо real-time агрегаций на миллионах строк

Вывод

Агрегация — это фундаментальная операция для аналитики и статистики. Она позволяет эффективно работать с большими объёмами данных, позволяя БД делать тяжёлую работу, а приложению — только получать результаты.