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

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

⚡ Шпаргалка

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

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

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

# JOIN
session.execute(select(User.name, Address.description).join(User.addresses)).all()
session.execute(
    select(User.name).outerjoin(User.addresses).where(Address.id.is_(None))
).all()

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

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

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

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

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

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

# where — явная запись условия
user = session.scalars(select(User).where(User.name == "Alice")).first()

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

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

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

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

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

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

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

# Обновление по имени
user = session.scalars(select(User).where(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.scalars(select(User).where(User.name == "Charlie")).first()
if user:
    session.delete(user)
    session.commit()

Агрегации

Функции func

from sqlalchemy import func, select

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

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

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

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

Подзапросы

from sqlalchemy import func, select

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

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

JOIN и связи

INNER JOIN

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

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

LEFT OUTER JOIN

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

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

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

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

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

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