Пишем свой SQL query builder на Python: DSL, кеширование в Redis и защита от инъекций

Я — 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 Tablesfrom query_tables.db import SQLiteQuerysqlite = 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, Orderingquery = 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.hiredFROM personJOIN (    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_addressLEFT JOIN (    SELECT employees.id, employees.ref_person, employees.ref_company, employees.hired     FROM employees) AS employees ON employees.ref_person = person.idWHERE person.id = %(person_id_1)s AND person.name LIKE %(person_name_2)sORDER 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. Моя библиотека делает именно это. Она оставляет кеш запроса, если по этим таблицам не было изменений.

# Вставка новой записи в addresstable['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