Использование агрегаций и подзапросов в SQLAlchemy

Агрегация — это процесс обработки данных для получения сводной информации. Ее можно сравнить с подведением итогов на собрании: вы собираете много данных (например, зарплаты сотрудников) и сводите их в одно значение (например, среднюю зарплату).

В SQL для этих целей используются агрегатные функции:

• SUM — сумма значений.
• AVG — среднее значение.
• COUNT — количество записей.
• MIN и MAX — минимальное и максимальное значение.

Пример SQL-запроса с использованием агрегатных функций:

SELECT department, COUNT(*) AS num_employees, AVG(salary) AS avg_salary
FROM employees
GROUP BY department;

Этот запрос подскажет вам, сколько сотрудников работает в каждом отделе и какова их средняя зарплата.

Реализация агрегаций в SQLAlchemy

SQLAlchemy предоставляет удобный интерфейс для выполнения агрегаций. Для этого используются методы и функции модуля sqlalchemy.sql.functions.

Начнем с простого, посчитаем количество пользователей в таблице User:

from sqlalchemy import func

# Предположим, вы уже создали сессию SQLAlchemy
# и таблицу User с полями id, name, и age.

user_count = session.query(func.count(User.id)).scalar()
print(f"Количество пользователей: {user_count}")

Функция func позволяет вызывать SQL-функции, такие как COUNT, SUM, AVG, и другие.

Теперь давайте узнаем, каков средний возраст наших пользователей:

average_age = session.query(func.avg(User.age)).scalar()
print(f"Средний возраст пользователей: {average_age}")

Обратите внимание на метод .scalar(), который возвращает первое значение результата.

Используем метод group_by, чтобы сгруппировать пользователей по имени и получить количество записей для каждого имени:

results = session.query(User.name, func.count(User.id).label('count')) \
    .group_by(User.name).all()

for name, count in results:
    print(f"Имя: {name}, Количество: {count}")

Метод label позволяет удобно называть результат функции, чтобы проще было работать с ним в дальнейшем.

---

Работа с подзапросами

Подзапросы (или субзапросы) — это запросы, которые используются внутри других запросов. Они позволяют извлекать данные, зависящие от результатов другого запроса.

Пример подзапроса на обычном SQL:

SELECT name, age
FROM employees
WHERE age > (SELECT AVG(age) FROM employees);

Этот запрос выбирает всех сотрудников, чей возраст превышает средний возраст сотрудников.

SQLAlchemy позволяет реализовывать подзапросы через объект subquery.

Создадим подзапрос для вычисления среднего возраста и используем его для фильтрации:

from sqlalchemy.orm import aliased

# Подзапрос для вычисления среднего возраста
subq = session.query(func.avg(User.age).label('avg_age')).subquery()

# Основной запрос использует подзапрос
results = session.query(User).filter(User.age > subq.c.avg_age).all()

for user in results:
    print(f"Имя: {user.name}, Возраст: {user.age}")

Здесь:

• subquery() делает подзапрос доступным для использования в основном запросе.
• subquery().c.<имя> позволяет обращаться к результату как к колонке.

Алиасы позволяют переименовывать таблицы или запросы для упрощения работы с подзапросами. Это полезно, когда в запросе участвуют несколько таблиц или один и тот же запрос используется дважды.

user_alias = aliased(User)

results = session.query(User.name, func.count(user_alias.id)) \
    .join(user_alias, User.id == user_alias.id) \
    .group_by(User.name).all()

---

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

Создадим небольшую базу данных и реализуем запросы разной сложности.

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

Base = declarative_base()

# Определим модели
class Department(Base):
    __tablename__ = 'departments'
    id = Column(Integer, primary_key=True)
    name = Column(String)

class Employee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
    salary = Column(Integer)
    department_id = Column(Integer, ForeignKey('departments.id'))

    department = relationship('Department')

# Создание базы данных SQLite
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# Добавление данных
dept1 = Department(name='IT')
dept2 = Department(name='HR')
session.add_all([dept1, dept2])
session.commit()

employees = [
    Employee(name='Alice', age=30, salary=80000, department=dept1),
    Employee(name='Bob', age=25, salary=50000, department=dept1),
    Employee(name='Charlie', age=35, salary=60000, department=dept2),
]
session.add_all(employees)
session.commit()

Задача 1: средняя зарплата по отделам

Используем group_by для подсчета средней зарплаты в каждом отделе:

results = session.query(Department.name, func.avg(Employee.salary).label('avg_salary')) \
    .join(Employee, Department.id == Employee.department_id) \
    .group_by(Department.name).all()

for dept, avg_salary in results:
    print(f"Отдел: {dept}, Средняя зарплата: {avg_salary}")

Задача 2: сотрудники с зарплатой выше среднего по их отделу

Создадим подзапрос для вычисления средней зарплаты и используем его:

subq = session.query(Employee.department_id, func.avg(Employee.salary).label('avg_salary')) \
    .group_by(Employee.department_id).subquery()

results = session.query(Employee.name, Employee.salary, Department.name) \
    .join(Department, Employee.department_id == Department.id) \
    .join(subq, Employee.department_id == subq.c.department_id) \
    .filter(Employee.salary > subq.c.avg_salary).all()

for name, salary, dept_name in results:
    print(f"Сотрудник: {name}, Зарплата: {salary}, Отдел: {dept_name}")

---

Советы по оптимизации агрегаций и подзапросов

1. Избегайте ненужных подзапросов. Некоторые подзапросы могут быть преобразованы в JOIN для повышения производительности.
2. Используйте индексы. Добавление индексов на столбцы, которые часто участвуют в агрегациях или фильтрах, значительно ускорит запросы.
3. Старайтесь группировать данные на уровне базы. Выполняйте агрегации ближе к базе данных, чтобы минимизировать объём извлекаемых данных.

from sqlalchemy import Index

Index('idx_salary', Employee.salary)

4. Профилируйте запросы. SQLAlchemy предоставляет методы для анализа времени выполнения запросов и их оптимизации.

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

INFO sqlalchemy.engine.Engine SELECT ...

Теперь вы готовы эффективно использовать агрегации и подзапросы для работы со сложными структурами данных, оптимизируя запросы и повышая производительность своего приложения.