Я — Python-разработчик, и большую часть времени работаю с Django и пишу сырые SQL запросы. Мне нравится Django ORM и я не имею ничего против, но разбираясь с legacy кодом, я невольно стал задумываться, что файлы содержащие модели огромны. Вроде ничего такого, так и должно быть, проект большой и много кода. Это в порядке вещей, так я решил. Выбросил лишние мысли и стал жить дальше. Но через несколько месяцев, мне показалось, что моделям не хватает кеширование данных. Порой кусок данных в несколько тысяц строк приходится получать не один раз. В какой-то момент я задался вопрос, что было бы, если бы в Django не было бы моделей. И в этот момент все завертелось. Это был мой вызов, который заставил меня двигаться. Мне хотелось видеть что-то похожее с SQLAlchemy или похожим на простой SQL, но при этом не контактировать с моделями и иметь возможность кешировать данные из коробки, а также иметь защиту от инъекций. Это был взрыв мозга, который зарядил меня делать эту библиотеку по праздникам, выходным и моим отпускам. Я был этим так увлечен, как ребенок в свой рождественский день, что не заметил, как быстро выросла библиотека со своей ссылкой на pypi.

Вот так появилась идея и библиотека, которую я назвал CORMless — «запросы в объектном стиле без моделей». Звучит, наверно, смешно, но давайте разберёмся, что это значит на практике.

Проблема: когда ORM избыточна

ORM — мощный инструмент. Но есть сценарии, где он становится обузой:

• Динамическая структура БД. Если у вас legacy-база или часто меняется структура БД, вы не можете заранее описать модели. Поддержка моделей превращается в бесконечную гонку. Что при этом придется делать: написать миграцию (к примеру, сервис миграций), обновить модель (возможно, в нескольких сервисах), согласовывать с командами, у которых есть зависимости по этому полю.

• Аналитика и сложные отчёты. Когда нужны тройные JOIN с HAVING и GROUP BY, а не CRUD для одной таблицы. Также ORM поддерживает не все фичи СУБД.

• Микросервисы-адаптеры. Часто нужно просто «сходить в БД, забрать данные и отдать их в другом формате». Заводить ради этого модели — как стрелять из пушки по воробьям.

Но писать сырой SQL руками — тоже не выход. Теряется объектность, появляется риск инъекций, код обрастает из f"SELECT * FROM {table}", а хотелось бы более менее объектный SDL.

В моей практике в Django-проектах в основном используются сырые SQL-запросы. ORM запросы тоже есть, но это если нужно выбрать запись из одной модели. Поэтому было желание уйти от сырых SQL запросов, но и не использовать модели.

Мне хотелось получить золотую середину: удобство цепочечного API, как в ORM, но без необходимости описывать классы моделей. И чтобы под капотом это был чистый, параметризованный SQL.

Архитектура: строим DSL на Python

Концепция проста: вы «просите» библиотеку прочитать структуру вашей БД, и она сама узнаёт, какие таблицы и поля существуют. Дальше вы строите запрос, как в Django ORM или SQLAlchemy Core:

from query_tables import Tables
from query_tables.db import SQLiteQuery

sqlite = SQLiteQuery('path/to/database.db')
table = Tables(sqlite, non_expired=True)  # включаем "вечный" кеш

# Простой запрос с фильтрацией
res = table['person'].filter(id=2).get()
# Вернёт список словарей: [{'person.id': 2, 'person.name': 'Anton 2', ...}]
# Наименование ключа - это название таблицы / alias таблицы, а через точку название поля.

Ключевое отличие от сырого SQL здесь — это объектное построение запроса. Методы .filter(), .join(), .select(), .group_by() не выполняются сразу. Они лишь собирают внутреннее представление запроса. А когда вы вызываете .get() — библиотека компилирует это в SQL, выполняет и возвращает результат.

Сложные JOIN без боли

Самой хардкорной частью разработки стала реализация вложенных JOIN. Я хотел, чтобы сложные запросы выглядели в коде так же структурированно, как и в голове у разработчика. Вот, например, как выглядит запрос с JOIN и фильтрацией по связанным таблицам:

from query_tables.query import Join, LeftJoin, AND, OR, Field, Ordering

query = table['person'].select(
    Field('person', 'id'), 
    Field('person', 'name'), 
    Field('person', 'age')
).join(
    Join(table['address'], Field('address', 'id'), Field('person', 'ref_address')).filter(
        OR( 
            AND(Field('address', 'street').like('%%ушкина'), Field('address', 'building').equ(10)),
            Field('address', 'building').in_([5, 10])
        )
    )
).join(
    LeftJoin(table['employees'], Field('employees', 'ref_person'), Field('person', 'id')).select(
        Field('employees', 'id'), 
        Field('employees', 'ref_person'), 
        Field('employees', 'ref_company'), 
        Field('employees', 'hired')
    )
).filter(
    Field('person', 'id').equ(1), Field('person', 'name').like('Ant%%')
).order_by(
    Field('person', 'age').desc()
)

res = query.get()

Класс Field здесь играет ключевую роль — он явно указывает, к какой таблице относится поле, и библиотека всегда может проверить, существует ли оно. Это убирает опечатки на этапе сборки запроса, а не на этапе выполнения в БД.

Библиотека сгенерирует такой SQL:

SELECT person.id, person.name, person.age, address.id, address.street, 
       address.building, employees.id, employees.ref_person, employees.ref_company, 
       employees.hired
FROM person
JOIN (
    SELECT address.id, address.street, address.building 
    FROM address
    WHERE ((address.street LIKE %(address_street_1)s AND address.building = %(address_building_2)s) 
           OR address.building IN (%(address_building_3)s, %(address_building_4)s))
) AS address ON address.id = person.ref_address
LEFT JOIN (
    SELECT employees.id, employees.ref_person, employees.ref_company, employees.hired 
    FROM employees
) AS employees ON employees.ref_person = person.id
WHERE person.id = %(person_id_1)s AND person.name LIKE %(person_name_2)s
ORDER BY person.age DESC

В сыром SQL запросе нет значений параметров, что предотвращает инъекций. Также из запроса можно увидеть, что каждая таблица подключается через подзапросы внутри JOIN. Это специально сделано, чтобы было визуальное соответствие с объектными запросами в стиле query_tables.

Кеширование: не просто «сохранить результат»

Отдельная большая тема — это кеширование. Просто взять и закешировать SQL-запрос недостаточно. Нужно знать, когда инвалидировать кеш.

Предположим, есть три запроса к БД:

• query1: получает данные из таблиц person + address

• query2: получает данные из таблиц person + address + employees + company

• query3: получает данные из таблиц person + employees

Данные закешированы. Теперь вы делаете INSERT в таблицу address. Что должно произойти? Кеш query1 и query2 должен быть сброшен, а query3 — нет, потому что он не зависит от address. Моя библиотека делает именно это. Она оставляет кеш запроса, если по этим таблицам не было изменений.

# Вставка новой записи в address
table['address'].insert(street='Новая', building=999)

# После этого кеш query1 и query2 будет автоматически очищен.
# query3 останется нетронутым.

Как это работает под капотом: для каждого кеша хранится отображение “хеш запроса → данные”. Плюс обратный индекс: “таблица → список хешей запросов”. Именно он позволяет быстро найти, какой кеш нужно сбросить при изменении конкретной таблицы. При любом изменении данных через INSERT, UPDATE или DELETE библиотека находит все хеши, связанные с изменяемой таблицей, и удаляет их из кеша. Это инвалидация по тегам (таблицам), реализованная вручную.

Поддерживается два типа кеша:

• In-memory: на основе aiocache или cachetools (TTLCache или LRUCache). Подходит для приложений, работающих в одном процессе.

• Redis: распределённый кеш, который могут использовать несколько экземпляров приложения одновременно.

В обоих случаях реализована защита от гонок.

С одной стороны можно было бы обойтись бы и одним видом кеша, скажем через Redis. Это было бы хорошим решением для Django. Но это стало бы ограничением для тех сервисов, которые не иcпользуют внешнее кеширование, а сохраняют значения в памяти процесса. Поэтому два вида кеша.

Синхронность и асинхронность: две параллельные вселенные

Python-экосистема сейчас раздвоена: есть синхронный мир (Django, Flask с WSGI) и асинхронный (FastAPI, aiohttp). Я хотел, чтобы библиотека работала в обоих мирах с одинаковым API.

Итоговая архитектура:

• Синхронная версия: SQLiteQuery, PostgresQuery, Tables, CacheQuery, RedisCache.

• Асинхронная версия: AsyncSQLiteQuery, AsyncPostgresQuery, TablesAsync, AsyncCacheQuery, AsyncRedisCache.

С точки зрения пользователя, код отличается только наличием await перед методами:

# Синхронный код
res = table['person'].filter(id=2).get()

# Асинхронный код
res = await async_table['person'].filter(id=2).get()

Под капотом пришлось построить параллельные иерархии классов. Самое сложное было — не допустить дублирования логики построения запросов. Поэтому она вынесена в общие модули (query/query.py, query/condition.py). Класс Query, отвечающий за компиляцию объектного DSL в SQL-строку, используется и синхронной, и асинхронной версией без каких-либо изменений. А вот специфика работы с БД (соединения, пулы) и кешем инкапсулирована в отдельных классах — QueryTable и AsyncQueryTable.

Безопасность: защита от инъекций на уровне библиотеки

Один из главных принципов при разработке — никакой конкатенации пользовательских данных в SQL. Вообще.

Библиотека использует параметризованные запросы. Все значения, которые вы передаёте в .filter() или .insert(), не вставляются напрямую в строку запроса. Вместо этого генерируется плейсхолдер %(field_name_N)s, а само значение уходит в словарь параметров:

# Ваш код:
table['person'].filter(name__like='%%Anton%%')

# Генерируется SQL:
# SELECT ... WHERE person.name LIKE %(person_name_1)s

# И параметры:
# {'person_name_1': '%%Anton%%'}

База данных сама экранирует параметры. SQL-инъекция становится невозможной на фундаментальном уровне. Вам не нужно помнить про экранирование — библиотека делает это за вас.

Функции, GROUP BY и всё остальное

Для полноты картины — библиотека поддерживает практически все операторы SQL, которые могут понадобиться в повседневной работе:

• Фильтрация: __like, __ilike, __in, __between, __gt/gte/lt/lte, __isnull, __regex, __iregex и их отрицания.

• SQL-функции: Max, Min, Avg, Count, Sum, Concat, Upper, Lower, Substring, Replace, Case, Coalesce, Extract, Interval и другие.

• Группировка и фильтрация после группировки: GROUP BY + HAVING.

• Сырые запросы: если возможностей DSL не хватает, всегда можно выполнить произвольный SQL через table.query('SELECT ...'), и его тоже можно кешировать.

Я не знал какие функции могут понадобиться другим, поэтому постарался добавить как можно больше возможностей со стороны CORMless.

Заключение

CORMless — это не замена SQLAlchemy или Django ORM. Это нишевый инструмент для ситуаций, когда модели избыточны, но писать сырой SQL не хочется.

Для меня этот проект стал отличной школой:

• Проектирования DSL на Python.

• Понимания, как работают ORM изнутри.

• Реализации сложной логики кеширования с инвалидацией.

• Построения двойных (синхронных и асинхронных) API без дублирования кода.

Конечно, еще есть куда развиваться и расти, но сейчас я доволен проделанной работой и хотел бы поделиться с вами своей идеей и своими результатами.

Ссылки

• Исходный код на GitHub: github.com/AntonGlyzin/query_tables

• Полная документация: query-tables.readthedocs.io

• Пакет на PyPI: pip install query-tables