← назад к разделу

Запрос в SQLAlchemy это объект, который собирают по частям и выполняют в сессии. Такой запрос можно передавать между функциями, дополнять фильтрами по условию и покрывать тестом без базы. Разберём, как строить запросы в стиле 2.0 так, чтобы SQL на выходе был предсказуем, а результат имел понятную форму.

Обязательно

select и четыре способа его выполнить

select() принимает то, что хотим получить: класс модели, отдельные колонки или выражения. От этого зависит форма результата.

from sqlalchemy import select, func

stmt = select(Order).where(Order.status == "PAID").order_by(Order.created_at.desc())

orders = session.scalars(stmt).all()           # список объектов Order
order = session.scalar(stmt.limit(1))          # один объект или None
rows = session.execute(select(Order.id, Order.total)).all()      # список Row, у строки .id и .total
mappings = session.execute(select(Order.id, Order.total)).mappings().all()   # список словарей

Правило чтения: scalars берёт первую колонку каждой строки, поэтому для select(Order) возвращает объекты; execute возвращает строки целиком. Если запрос с двумя колонками выполнить через scalars, вторая молча потеряется, и это частая первая ошибка.

Для одного объекта по первичному ключу есть session.get(Order, 42): он сначала смотрит в identity map и не ходит в базу, если объект уже загружен. Для одного объекта с гарантией существования session.scalars(stmt).one() поднимет исключение, если строк ноль или больше одной.

Фильтры, которые не ведут себя как в Python

where принимает выражения на колонках, и операторы перегружены: ==, !=, <, in_, like, ilike, between. Логику соединяют and_, or_ или перечислением аргументов в where (это AND). Два места, где интуиция Python подводит:

Order.deleted_at == None компилируется в IS NULL, а не в = NULL, так задумано, но линтеры ругаются; явный Order.deleted_at.is_(None) читается честнее.

Order.id.in_([]) с пустым списком не ошибка: SQLAlchemy компилирует выражение так, что оно всегда ложно, и запрос вернёт пусто. Это удобно, но скрывает логическую ошибку выше по коду, когда список пустой случайно.

Фильтры удобно собирать по условию, потому что where возвращает новый запрос:

def list_orders(session, *, status: str | None, customer_id: int | None, limit: int):
    stmt = select(Order)
    if status is not None:
        stmt = stmt.where(Order.status == status)
    if customer_id is not None:
        stmt = stmt.where(Order.customer_id == customer_id)
    return session.scalars(stmt.order_by(Order.id).limit(limit)).all()

Соединения и агрегаты

join по объявленной связи не требует условия: SQLAlchemy берёт его из relationship или внешнего ключа. Агрегаты пишут через func и группируют group_by:

stmt = (
    select(Customer.name, func.count(Order.id).label("orders"), func.sum(Order.total).label("revenue"))
    .join(Order, Order.customer_id == Customer.id)
    .where(Order.status == "PAID")
    .group_by(Customer.id, Customer.name)
    .having(func.count(Order.id) >= 3)
    .order_by(func.sum(Order.total).desc())
)
for row in session.execute(stmt):
    print(row.name, row.orders, row.revenue)

Это Core-запрос внутри ORM-сессии: на выходе строки, не объекты, и это правильно для отчёта. outerjoin даёт LEFT OUTER JOIN; func.coalesce(func.sum(...), 0) спасает от None у клиентов без заказов. Любая функция базы доступна как func.имя, SQLAlchemy не проверяет её существование, это сделает PostgreSQL при выполнении.

Подзапросы и CTE

Подзапрос это тот же select, превращённый в .subquery() или .scalar_subquery(). Типичная задача «последний заказ каждого клиента»:

last = (
    select(Order.customer_id, func.max(Order.created_at).label("last_at"))
    .group_by(Order.customer_id)
    .subquery()
)
stmt = select(Customer.name, last.c.last_at).join(last, last.c.customer_id == Customer.id)

Колонки подзапроса доступны через .c. Для рекурсии (дерево категорий) и для читаемости больших запросов берут .cte(), который компилируется в WITH. Статья про JSONB и рекурсивные запросы в PostgreSQL показывает такие случаи со стороны базы.

Оконные функции: func.row_number().over(partition_by=Order.customer_id, order_by=Order.created_at.desc()). Ими часто заменяют коррелированные подзапросы, и план получается проще.

text: когда SQL честнее

Когда выражение на Python длиннее и темнее самого SQL, пишут text() с именованными параметрами. Параметры передают отдельно, в строку их никогда не подставляют:

from sqlalchemy import text

stmt = text("""
    SELECT date_trunc('week', created_at) AS week, sum(total) AS revenue
    FROM orders
    WHERE created_at >= :since AND status = 'PAID'
    GROUP BY 1 ORDER BY 1
""")
rows = session.execute(stmt, {"since": since}).mappings().all()

:since это параметр драйвера, и даже если since пришёл от пользователя, инъекции не будет. Инъекция появляется ровно в одном случае: f-строка с данными внутри SQL. Если нужно подставить имя колонки для сортировки, его проверяют по белому списку и собирают через getattr(Order, name), а не строкой. Подробнее об этом в статье про проверки на границе.

text() можно сочетать с ORM: select(Order).from_statement(text(...)) соберёт объекты из строк готового SQL, если он возвращает нужные колонки.

Постраничная выборка

OFFSET для глубоких страниц дорог: база читает и выбрасывает все предыдущие строки. Для бесконечных лент берут курсор по сортируемому ключу:

def page_orders(session, *, after_id: int | None, limit: int):
    stmt = select(Order).order_by(Order.id)
    if after_id is not None:
        stmt = stmt.where(Order.id > after_id)
    return session.scalars(stmt.limit(limit)).all()

Клиенту отдают идентификатор последней строки как курсор. При сортировке по неуникальному полю (дате) ключ составной: (created_at, id), и условие пишут через кортежное сравнение tuple_(Order.created_at, Order.id) > (after_at, after_id). Со стороны API это описано в статье про параметры запроса.

Массовые операции без объектов

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

from sqlalchemy import update, insert, delete

result = session.execute(
    update(Order).where(Order.status == "NEW", Order.created_at < cutoff).values(status="EXPIRED")
)
print(result.rowcount)

session.execute(insert(Order), [{"customer_id": 1, "total": 10}, {"customer_id": 2, "total": 20}])

insert(Model) со списком словарей это пакетная вставка одним выражением с многими VALUES, в разы быстрее session.add в цикле (на пяти тысячах строк разница около трёхкратной) и без объектов в сессии; подробности в статье про производительность. Важно помнить: такие выражения обходят объекты в сессии. Если заказ уже загружен и вы обновили его статус через update(), объект в памяти об этом не знает; параметр execution_options(synchronize_session="fetch") просит сессию согласовать состояние, но проще не смешивать два стиля в одной единице работы.

Дополнительно: при первом чтении можно пропустить

Глубже: запрос как значение и спецификациирасширенное

Поскольку select() это объект, фильтры можно хранить и комбинировать отдельно от выполнения. Репозиторий принимает не десяток именованных аргументов, а условие:

from sqlalchemy import ColumnElement

def paid_after(since) -> ColumnElement[bool]:
    return and_(Order.status == "PAID", Order.created_at >= since)

def find(session, *criteria: ColumnElement[bool], limit: int = 100):
    return session.scalars(select(Order).where(*criteria).order_by(Order.id).limit(limit)).all()

Это дешёвая форма паттерна «спецификация»: условия именованы, тестируются компиляцией в SQL без базы (str(stmt.compile(compile_kwargs={"literal_binds": True}))) и переиспользуются. Граница у приёма одна: условия должны оставаться внутри слоя хранения. Как только ColumnElement просачивается в домен или в API, модель данных утекает наружу, и переименование колонки ломает обработчики. В гексагональной раскладке репозиторий принимает доменные фильтры (значения и перечисления), а в выражения SQLAlchemy их переводит сам.

Второй приём для чтения: select(...).execution_options(populate_existing=True) перечитывает уже загруженные объекты свежими данными, что бывает нужно после массового update() в той же сессии.

Коротко

  • select(Model) выполняют через scalars (объекты), select(колонки) через execute (строки или mappings()); scalars на нескольких колонках молча теряет все, кроме первой.
  • == None даёт IS NULL (лучше is_(None)), in_([]) всегда ложно и возвращает пусто; where возвращает новый запрос, фильтры собирают по условию.
  • Соединения по связи без условия, агрегаты через func и group_by; результат отчёта это строки, не объекты.
  • Подзапросы через .subquery() с доступом к .c, рекурсия и читаемость через .cte(), оконные функции через .over().
  • text() с именованными параметрами безопасен, f-строка с данными в SQL нет; имена колонок только по белому списку.
  • Глубокие страницы через курсор по ключу, а не OFFSET; при неуникальной сортировке ключ составной.
  • Массовые update(), insert() со списком и delete() обходят объекты сессии; не смешивать с загруженными объектами в одной единице работы.

Что почитать дальше