SQLAlchemy — очень удобный инструмент. В нём явно видно, какой select() строится, где используются join, какие условия попадают в where, как загружаются связи и какой SQL в итоге уходит в базу.
Но в обычном backend-коде далеко не каждый запрос является сложным.
Во многих Flask/FastAPI-сервисах есть большое количество однотипных запросов: отфильтровать записи, пройти по связи, добавить OR, отсортировать результат, ограничить количество строк, заранее загрузить связанные объекты и вернуть список результатов.
Такой код часто выглядит примерно так:
stmt = ( select(Item) .join(Item.group) .where( or_( Item.is_valid == True, Item.number > 100, ) ) .where(Group.is_active == True) .options(joinedload(Item.group)) .order_by(Item.number.desc()) .limit(20))items = session.execute(stmt).scalars().all()
Проблема: бизнес-смысл запроса заметно короче, чем код вокруг него.
Фактически мы хотим сказать следующее:
Дай мне валидные Item или Item с number > 100,только из активных групп,сразу загрузи group,отсортируй по number по убываниюи верни 20 строк.
После нескольких десятков похожих запросов в приложении начинает хотеться более компактного слоя, который оставляет SQLAlchemy в основе, но убирает часть повторяющейся механики.
Обычно такой слой всё равно появляется в проекте, даже если его никто специально не проектировал. Сначала это несколько helper-функций для фильтров, потом общий код для сортировки и пагинации и так далее. В какой-то момент становится понятно, что уже написан собственный мини-query layer.
В результате похожей истории я решил вынести этот слой в отдельный пакет. Так и появился sqlalchemy-query-manager.
Идея очень простая: добавить поверх SQLAlchemy небольшой query layer с Django-style фильтрами, Q-объектами, relationship lookups, eager loading и preview итогового SQL.
Это вспомогательный слой для типовых backend-запросов, где прямой SQLAlchemy-код начинает выглядеть слишком громоздким.
Какой API хотелось получить
Тот же запрос можно записать так:
items = ( Item.query_manager .where( Q(is_valid=True) | Q(number__gt=100), group__is_active=True, ) .select_related("group") .order_by("-number") .limit(20) .all())
Здесь остаётся тот же смысл:
-
Q(is_valid=True) | Q(number__gt=100)описывает условиеOR; -
group__is_active=Trueфильтрует по связанной модели; -
select_related("group")явно говорит, что связь нужно загрузить заранее; -
order_by("-number")сортирует по убыванию; -
limit(20)ограничивает результат; -
.all()выполняет запрос.
Главная цель такого API — приблизить код к намерению, не теряя SQLAlchemy как основу. Такой слой может уменьшить количество повторяющегося кода.
Основа query-manager
query-manager привязывается к SQLAlchemy-модели и работает как chainable query builder.
Пример:
items = ( Item.query_manager .where(is_valid=True) .where(number__gte=100) .order_by("-number") .limit(20) .all())
Каждый вызов добавляет состояние будущего запроса. Сам запрос не выполняется сразу после .where() или .order_by().
Выполнение происходит только в конце при:
.all().first().get(...).count().exists(...)
То есть manager не пытается неявно ходить в базу на каждом шаге. Он собирает запрос, а затем выполняет его в конце цепочки.
Lookup-строки вместо повторяющихся выражений
Самая полезная часть для меня — упрощение написания условий в запросе.
Вместо такого условия:
Item.number >= 100
можно написать:
number__gte=100
Вместо ручного join и фильтра по связанной модели:
select(Item).join(Item.group).where(Group.is_active == True)
можно написать:
group__is_active=True
Ключ group__is_active несёт в себе путь:
group__is_active
Это можно прочитать так:
relationship: groupfield: is_activeoperator: equality
Если путь глубже, он разбирается так же:
group__owner__email__isnull=False
Здесь смысл уже такой:
relationship path: group -> ownerfield: emailoperator: isnull
В обычном SQLAlchemy это тоже можно сделать без проблем. Но каждый раз нужно явно писать переходы по relationship, join и условиям. В CRUD-heavy коде такая механика быстро начинает повторяться.
Lookup-строки как раз забирают на себя эту рутинную часть.
Примеры операторов
Простейший случай — equality:
Item.query_manager.where(is_valid=True).all()
Для сравнений можно использовать суффиксы:
Item.query_manager.where(number__gt=100).all()Item.query_manager.where(number__gte=100).all()Item.query_manager.where(number__lt=500).all()Item.query_manager.where(number__lte=500).all()
Для проверки на NULL:
Item.query_manager.where(name__isnull=False).all()
Для фильтрации по списку:
Item.query_manager.where(number__in=[1, 2, 3]).all()
Для строковых условий можно использовать like и ilike:
Item.query_manager.where(name__like="test%").all()Item.query_manager.where(name__ilike="test%").all()
Идея в том, чтобы часто используемые фильтры можно было записывать компактно и одинаково во всём приложении.
Q-объекты для составных условий
Keyword-фильтры естественно складываются через AND:
items = ( Item.query_manager .where( is_valid=True, number__gte=100, ) .all())
Это читается как:
is_valid = true AND number >= 100
Но в реальном коде часто нужен OR.
Для этого используются Q-объекты:
items = ( Item.query_manager .where( Q(is_valid=True) | Q(number__gt=100) ) .all())
Можно группировать условия:
items = ( Item.query_manager .where( (Q(is_valid=True) | Q(number__gt=100)) & Q(group__is_active=True) ) .all())
Такой код читается почти как само условие:
(is_valid = true OR number > 100) AND group.is_active = true
SQLAlchemy уже умеет делать это через or_() и and_():
where( and_( or_( Item.is_valid == True, Item.number > 100, ), Group.is_active == True, ))
Q-объекты полезны там, где условия собираются динамически: из query parameters, фильтров в API, формы поиска или внутренних правил бизнес-логики.
Фильтры по relationship
Отдельный случай — фильтрация по связанным моделям.
Например, есть Item, у которого есть relationship group, а у Group есть поле is_active.
В чистом SQLAlchemy мы обычно пишем что-то вроде:
stmt = ( select(Item) .join(Item.group) .where(Group.is_active == True))
Через manager это можно записать так:
items = ( Item.query_manager .where(group__is_active=True) .all())
Если связь вложенная, путь продолжается:
items = ( Item.query_manager .where(group__owner__email__isnull=False) .all())
Внутри manager разбирает lookup key по __, определяет путь по связанным моделям, находит конечное поле и строит соответствующее условие.
Упрощённо процесс выглядит так:
group__owner__email__isnull=False
превращается в:
пройти от Item к group;пройти от Group к owner;взять поле Owner.email;применить оператор isnull;добавить нужные JOIN-ы и условие в WHERE.
Это один из тех случаев, где абстракция действительно упрощает работу. Особенно если в проекте много фильтров по связанным сущностям.
Eager loading должен быть виден в коде запроса
Ещё одна вещь, которую я хотел оставить явной, — это eager loading.
В web-приложениях session scope часто короткий. Кроме того, при работе через sessionmaker или context manager пакет после операции может отдавать уже detached-объекты. В такой ситуации обращение к lazy relationship после выполнения запроса легко приводит к DetachedInstanceError.
Поэтому загрузка связей должна быть видна прямо в query chain:
items = ( Item.query_manager .select_related("group") .all())
Или так:
items = ( Item.query_manager .prefetch_related("group") .all())
Названия вдохновлены Django, но цель здесь практическая: сделать загрузку relationship явной на уровне кода. Внутри select_related() использует JOIN-based loading, а prefetch_related() — отдельную загрузку через selectinload.
Когда в запросе написано:
.select_related("group")
сразу понятно, что group должен быть загружен вместе с Item.
SQL preview
Любая обёртка над базой данных становится опасной, если разработчик перестаёт понимать, какой SQL она генерирует.
Поэтому в manager есть возможность посмотреть итоговый SQL:
sql = ( Item.query_manager .where( Q(is_valid=True) | Q(number__gt=100), group__is_active=True, ) .get_sql_query())print(sql)
Для меня это было важно так как очень часто приходится смотреть на реальные SQL запросы при правке багов или написания нового функционала.
CRUD helpers
В пакете также есть базовые CRUD helpers:
Item.query_manager.create(...)Item.query_manager.bulk_create(...)Item.query_manager.where(id=1).update(...)Item.query_manager.where(is_valid=False).delete()
Например, update привязан к фильтру:
Item.query_manager.where(id=1).update(name="updated")
То же самое с delete:
Item.query_manager.where(is_valid=False).delete()
Мне кажется, это нормальная форма. Сначала явно задаётся набор строк, к которому применяется операция, и только потом вызывается update() или delete().
Aggregates и count
Для обычных backend-задач часто нужны простые запросы на подсчёт строк, проверку наличия записи или получение агрегированных значений.
Примеры:
count = Item.query_manager.where(is_valid=True).count()
exists = Item.query_manager.where(number__gt=100).exists()
Для агрегатов есть отдельный метод aggregate():
from sqlalchemy_query_manager.core.helpers import Sum, Count, Avg, Min, Maxstats = Item.query_manager.where(is_valid=True).aggregate( total=Sum("number"), count=Count("id"), avg=Avg("number"), min=Min("number"), max=Max("number"),)
Такие вещи можно писать и через SQLAlchemy напрямую. Но если они часто встречаются рядом с остальными фильтрами, удобно иметь их в том же стиле.
Основной компромисс
У такого подхода есть понятный trade-off.
Мы выигрываем:
- более короткую запись типовых запросов;- Django-style lookup notation;- composable Q-фильтры;- фильтрацию по relationship path;- явный eager loading в query chain;- меньше повторяющегося session/query boilerplate;- возможность быстро посмотреть итоговый SQL.
Но теряем:
- часть явности;- часть прямого контроля в каждой строке кода;- не все функции SQLAlchemy могу поддерживаться
Мне кажется, на данный момент такой слой имеет смысл в проектах, где много обычных Flask/FastAPI endpoint-ов с фильтрацией, сортировкой, limit/offset, relationship filters и простыми CRUD-операциями.
Итог
sqlalchemy-query-manager — это небольшой слой поверх SQLAlchemy для случаев, где обычные backend-запросы становятся слишком большими.
Код пакета можно посмотреть на GitHub:
https://github.com/ViAchKoN/sqlalchemy-query-manager
Установить можно используя pip:
pip install sqlalchemy-query-manager
Очень буду благодарен за обратную связь, так как планирую дорабатывать пакет далее.
ссылка на оригинал статьи https://habr.com/ru/articles/1052518/