|
Professor Seleznov
|
Я — 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 без дублирования кода.
Конечно, еще есть куда развиваться и расти, но сейчас я доволен проделанной работой и хотел бы поделиться с вами своей идеей и своими результатами.
-
Ссылки
-Источник
|
