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: group
field:        is_active
operator:     equality

Если путь глубже, он разбирается так же:

group__owner__email__isnull=False

Здесь смысл уже такой:

relationship path: group -> owner
field:             email
operator:          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, Max

stats = 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

Очень буду благодарен за обратную связь, так как планирую дорабатывать пакет далее.