🔖 Справочник: SQLAlchemy-запросы

Шпаргалка для практикума — все методы в одном месте

⚡ Шпаргалка

# Чтение
session.query(User).filter(User.name == "Alice").first()
session.query(User).filter(User.age > 20).all()
session.get(User, user_id)
session.query(User).order_by(User.age.desc()).all()
session.query(User).order_by(User.name).limit(4).all()
session.query(session.query(User).filter_by(name="X").exists()).scalar()

# Запись / изменение / удаление
session.add(User(name="X", age=0)); session.commit()
user.age = 25; session.commit()
session.delete(user); session.commit()

# Агрегации
from sqlalchemy import func
session.query(func.avg(User.age)).scalar()
session.query(func.max(User.age)).scalar()
session.query(User.age, func.count(User.id)).group_by(User.age).all()
.having(func.count(User.id) > 1)
subq = session.query(func.avg(User.age)).subquery()

# JOIN
session.query(User.name, Address.description).join(User.addresses).all()
session.query(User.name).outerjoin(User.addresses).filter(Address.id.is_(None)).all()

Базовые запросы

Чтение данных

# Найти первого пользователя с заданным именем
user = session.query(User).filter(User.name == "Alice").first()

# Все пользователи с условием
users = session.query(User).filter(User.age > 20).all()

# Несколько полей
rows = session.query(User.name, User.age).filter(User.age < 30).all()

# Найти по первичному ключу (возвращает None, если не найдено)
user = session.get(User, user_id)

# filter_by — краткая запись для сравнения по равенству
user = session.query(User).filter_by(name="Alice").first()

Сортировка и ограничение

# По убыванию возраста
session.query(User.name, User.age).order_by(User.age.desc()).all()

# По алфавиту, первые 4
session.query(User.name).order_by(User.name).limit(4).all()

Проверка существования

# Возвращает True/False
exists = session.query(
    session.query(User).filter_by(name="Charlie").exists()
).scalar()

Создание / изменение / удаление

# Добавление
new_user = User(name="Charlie", age=40)
session.add(new_user)
session.commit()

# Обновление по имени
user = session.query(User).filter(User.name == "Bob").first()
if user:
    user.age = 25
    session.commit()

# Обновление по id
user = session.get(User, 5)
if user:
    user.age = 35
    session.commit()

# Удаление
user = session.query(User).filter(User.name == "Charlie").first()
if user:
    session.delete(user)
    session.commit()

Агрегации

Функции func

from sqlalchemy import func

# Среднее
result = session.query(func.avg(User.age)).scalar()

# Максимум и минимум
max_age = session.query(func.max(User.age)).scalar()
min_age = session.query(func.min(User.age)).scalar()

# Группировка с подсчётом
age_groups = (
    session.query(User.age, func.count(User.id))
    .group_by(User.age)
    .all()
)

# Фильтрация групп через having
age_groups = (
    session.query(User.age, func.count(User.id).label('cnt'))
    .group_by(User.age)
    .having(func.count(User.id) > 1)
    .all()
)

Подзапросы

from sqlalchemy import func

# Создать подзапрос
subq = session.query(func.avg(User.age).label('avg_age')).subquery()

# Использовать в основном запросе
users = session.query(User).filter(
    User.age > subq.c.avg_age
).all()

JOIN и связи

INNER JOIN

# Пользователи вместе с адресами (только те, у кого есть адрес)
rows = (
    session.query(User.name, Address.description)
    .join(User.addresses)
    .all()
)

# Пользователи из конкретного города
users = (
    session.query(User.name)
    .join(User.addresses)
    .filter(Address.description == "Berlin")
    .all()
)

LEFT OUTER JOIN

# Пользователи БЕЗ адресов (Address.id IS NULL)
users = (
    session.query(User.name)
    .outerjoin(User.addresses)
    .filter(Address.id.is_(None))
    .all()
)

Агрегация через JOIN

# Количество пользователей в каждом городе
from sqlalchemy import func

counts = (
    session.query(Address.description, func.count(User.id))
    .join(Address.user)
    .group_by(Address.description)
    .all()
)

Обновление через relationship

# Изменить адрес конкретного пользователя
user = session.query(User).filter(User.name == "Bob").one_or_none()
if user and user.addresses:
    user.addresses[0].description = "Paris"
    session.commit()