Победа над ORM путем кодогенерации

Сразу хочу отметить перед читателем, что это не просто вольные рассуждения на тему, а в том числе и презентация моей библиотеки для Python, которую можно найти на github и установить через pip, и которая трудится в моей многопользовательской игре как SQL движок проекта.

Вступление

Проблемы ORM известны всем, кто хоть раз ими пользовался. Об этом существует множество статей как у нас (1, 2, 3, 4), так и в зарубежных источниках (1, 2, 3, 4). Эти проблемы в общем можно объединить довольно сложным термином Object-relational impedance mismatch, что позволю себе вольно перевести как "Объектно-реляционная разница потенциалов".

Альтернативой использованию ORM всегда было использование чистых драйверов баз данных и написание сырых SQL запросов, которые в свою очередь очень тяжело поддерживать и рефакторить в реальных проектах.

В этой статье я не буду хаить ORM (до меня это уже делали на протяжении без малого полуторадесятка лет), но хочу предложить альтернативный путь к решению задачи доступа к базе данных из потенциально любого другого ЯП.

Итак, вот основные постулаты, от которых я отталкивался при проектировании моей библиотеки:

1. Писать запросы будем на чистом SQL. Без промежуточного DSL, без ООП-оберток, без SQL-билдеров.

2. Вызывать запросы будем как стандартную функцию.

3. Аргументы функции - это аргументы SQL запроса.

4. Вставлять аргументы в нужные места запроса будем напрямую при помощи f-string. При этом не должно быть никаких SQL инъекций: вставку аргументов отдаем на откуп выбранному драйверу базы данных.

5. Возвращаемое значение функции - это всегда строка, кортеж (row) или массив кортежей определенного типа данных. Исключение: одномерные данные, когда возвращаем значение или массив значений.

6. У всех функций должны быть известны и явно определены в коде программы входные типы аргументов и выходные типы данных. Таким образом линтер станет на нашу сторону и будет помогать нам в разработке и рефакторинге.

Очевидно, что пункты 5 и 6 вынуждают нас либо прописывать типы данных и аргументов явно в коде самостоятельно, либо искать возможность их автоматической генерации. Таким образом мы становимся на путь кодогенерации на основе текущей схемы базы данных.

Какие вещи не будут реализованы, но есть в ORM:

1. Транспиляция на разные SQL диалекты. Существуют отдельные решения, которые транспилируют код из одного SQL диалекта в другой.

2. Управление транзакциями. ORM предоставляют обобщенный для разных драйверов интерфейс управления транзакциями (например, объект sqlalchemy.orm.Session). Моя же библиотека только делает запросы, а транзакциями пусть занимается драйвер.

querky

Так родилась моя библиотека, которую я назвал querky (github, pypi). Само слово - это совмещение английских слов query ("запрос") и созвучного quirky ("необычный", "причудливый").

Установить ее можно через pip командой pip install querky

Вот пример скелета проекта, который можно запустить и покрутить в руках возможности кодогенерации. Все примеры взяты из README проекта.

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

Теперь же я обязан огорчить многих моих читателей, отметив, что на данном этапе библиотека может работать только с asynpg.

Можно довольно просто сделать обертки для остальных PostgreSQL Python драйверов: psycopg2, psycopg (aka psycopg3) и aiopg. Однако возможно ли реализовать type-inference в остальных RDBMS - об этом я пока не зарекаюсь.

Конфигурация проекта

Установка

Устанавливаем библиотеку с драйвером asyncpg: pip install querky[asyncpg]

Структура проекта

Названия файлов и папок, а также общая структура проекта предоставляются лишь в качестве примера, и не являются единственной возможной конфигурацией проекта.

Создаем в корне проекта два файла querky_def.py и querky_gen.py, а также папку sql, в которой будем создавать .py файлы с SQL запросами. В папке sql создадим также пустой файл с названием __init__.py, чтобы сделать папку модулем.

Структура проекта на данный момент:

sql
|__ __init__.py
querky_def.py
querky_gen.py

querky_def

Файл настройки кодогенерации. В нем хранится единственный объект qrk, который мы будем использовать для объявления Python функций SQL запросами.

import os
from querky.presets.asyncpg import use_preset


qrk = use_preset(os.path.dirname(__file__), type_factory='dataclass+slots')

Для простоты примера используем пресет. Его хватит для 95% случаев, однако если нужна точечная настройка, то рекомендую использовать объект Querky напрямую. Подробнее в документации.

• Первый аргумент должен быть путем к корню вашего проекта.

• type_factory - это аргумент, определяющий какие классы будут генерироваться для результатов SQL запросов. По-умолчанию это typed_dict, но также можно использовать собственные типы, о чем подробнее есть в документации.

• Остальные аргументы такие же как у конструктора объекта Querky. О них подронее также в документации.

querky_gen

Этот скрипт отвечает непосредственно за генерацию кода. Мы будем запускать его, чтобы сгенерировать новые функции-обертки для запросов и перегенерировать устаревшие типы и сигнатуры.

Снова используем пресет:

import asyncio

from querky.presets.asyncpg import generate

from querky_def import qrk
import sql


if __name__ == "__main__":
    asyncio.run(generate(qrk, "<ваш url для подсоединения к postgres>", base_modules=(sql, )))

Как видно, импортируем базовый модуль sql и помещаем его в кортеж base_modules. При запуске скрипт пройдется по всем .py файлам внутри папки sql , собирая все SQL функции и генерируя для каждой из них обертку. Все сгенерированные функции и типы будут помещены в подпапке queries в файл с тем же названием, что исходный.

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

Схема базы данных

База данных: PostgreSQL 14.10

CREATE TABLE account (
    id BIGSERIAL PRIMARY KEY,
    username TEXT UNIQUE NOT NULL,
    first_name TEXT NOT NULL,
    last_name TEXT,
    phone_number TEXT,
    balance BIGINT NOT NULL DEFAULT 0,
    join_ts TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    referred_by_account_id BIGINT REFERENCES account (id)
);

CREATE TABLE post (
    id BIGSERIAL PRIMARY KEY,
    poster_id BIGINT NOT NULL REFERENCES account (id),
    message TEXT NOT NULL,
    ts TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE post_comment (
    id BIGSERIAL PRIMARY KEY,
    post_id BIGINT NOT NULL REFERENCES post (id),
    commenter_id BIGINT NOT NULL REFERENCES account (id),
    message TEXT NOT NULL,
    ts TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Написание запросов

Все, что нужно сделать, для объявления обычной функции запросом - это импортировать объект qrk, объявленный в querky_def.py , и использовать его метод query в качестве декоратора.

Например, создадим в папке sql файл example.py для наших запросов. Теперь структура проекта будет выглядеть так:

sql
|__ __init__.py
|__ example.py
querky_def.py
querky_gen.py

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

from querky_def import qrk


# запрос ничего не возвращает
@qrk.query  # или явно: @qrk.query(shape='status')
def update_account_phone_number(account_id, new_phone_number):
    return f'''
        UPDATE
            account
        SET
            phone_number = {+new_phone_number}
        WHERE
            id = {+account_id}
        '''


# запрос возвращает одно обязательное скалярное значение
@qrk.query(shape='value', optional=False)
def insert_account(username, first_name, last_name, phone_number, balance, referred_by_account_id):
    return f'''
        INSERT INTO
            account
            (
                username,
                first_name,
                last_name,
                phone_number,
                balance,
                referred_by_account_id
            )
        VALUES
            (
                {+username},
                {+first_name},
                {+last_name},
                {+phone_number},
                {+balance},
                {+referred_by_account_id}
            )
        RETURNING
            id
        '''


# запрос возвращает массив скалярных значений
@qrk.query(shape='column')
def select_top_largest_balances(limit):
    return f'''
        SELECT
            balance
        FROM
            account
        ORDER BY
            balance DESC
        LIMIT
            {+limit}
        '''

# запрос возвращает одно значение типа AccountReferrer или None (optional=True)
@qrk.query('AccountReferrer', shape='one', optional=True)
def get_account_referrer(account_id):
    return f'''
        SELECT
            referrer.id,
            referrer.username,
            referrer.first_name,
            referrer.last_name,
            referrer.join_ts

        FROM 
            account

        INNER JOIN
            account AS referrer
        ON
            account.referred_by_account_id = referrer.id

        WHERE
            account.id = {+account_id}
        '''


# запрос возвращает массив значений типа AccountPostComment
@qrk.query('AccountPostComment', shape='many')
def select_last_post_comments(post_id, limit):
    return f'''
        SELECT 
            account.first_name,
            account.last_name,
            post_comment.id,
            post_comment.message

        FROM
            post_comment

        INNER JOIN
            account
        ON
            post_comment.commenter_id = account.id

        WHERE
            post_comment.post_id = {+post_id}

        ORDER BY
            post_comment.ts DESC

        LIMIT
            {+limit}
        '''

Обратите внимание, что аргументы внутри запроса отмечены знаком +.

• Первым аргументом декоратора будет название типа, если применимо.

  • В случае скалярных результатов указание типа надпишет сгенерированную аннотацию.

  • В случае кортежей при отсутствии явного названия типа данных название будет сгенерировано из CamelCase-отформатированного названия функции.

• Флаг optional указывает на то, может ли функция вернуть None.

• Также необходимо явно указать форму возвращаемого значения.

  • status - по-умолчанию. Возвращает строковый статус выполнения запроса.

  • value - возвращает одно скалярное значение. Поддерживает флаг optional.

  • column - возвращает массив скалярных значений. Поддерживает флаг optional для элементов.

  • one - возвращает один кортеж. Поддерживает флаг optional.

  • many - возвращает массив кортежей.

С учетом конфигурации эти функции уже можно вызывать, если по какой-то причине хочется пропустить процедуру кодогенерации. querky займется подстановкой аргументов, а возвращаемые типы будут зависеть от используемых настроек драйвера базы данных.

Но мы пойдем дальше и запустим скрипт querky_gen.py. Если все прошло успешно, мы получим новый файл с названием example.py в подкаталоге sql/queries. Если папки queries не существует, она будет создана автоматически.

Новая структура проекта:

sql
|__ __init__.py
|__ example.py
|__ queries
    |__ example.py
querky_def.py
querky_gen.py

Давайте заглянем внутрь сгенерированного файла sql/queries/example.py:

# ~ AUTOGENERATED BY QUERKY ~ #
import datetime
from dataclasses import dataclass
from asyncpg import Connection
from sql.example import update_account_phone_number as _q0
from sql.example import insert_account as _q1
from sql.example import select_top_largest_balances as _q2
from sql.example import get_account_referrer as _q3
from sql.example import select_last_post_comments as _q4


async def update_account_phone_number(__conn: Connection, /, account_id: int, new_phone_number: str) -> str:
    return await _q0.execute(__conn, account_id, new_phone_number)


async def insert_account(__conn: Connection, /, username: str, first_name: str, last_name: str, phone_number: str, balance: int, referred_by_account_id: int) -> int:
    return await _q1.execute(__conn, username, first_name, last_name, phone_number, balance, referred_by_account_id)


async def select_top_largest_balances(__conn: Connection, /, limit: int) -> list[int]:
    return await _q2.execute(__conn, limit)


@dataclass(slots=True)
class AccountReferrer:
    id: int
    username: str
    first_name: str
    last_name: str
    join_ts: datetime.datetime


async def get_account_referrer(__conn: Connection, /, account_id: int) -> AccountReferrer | None:
    return await _q3.execute(__conn, account_id)

_q3.bind_type(AccountReferrer)


@dataclass(slots=True)
class AccountPostComment:
    first_name: str
    last_name: str
    id: int
    message: str


async def select_last_post_comments(__conn: Connection, /, post_id: int, limit: int) -> list[AccountPostComment]:
    return await _q4.execute(__conn, post_id, limit)

_q4.bind_type(AccountPostComment)


__all__ = [
    "select_top_largest_balances",
    "select_last_post_comments",
    "AccountReferrer",
    "get_account_referrer",
    "AccountPostComment",
    "update_account_phone_number",
    "insert_account",
]

Как видите, мы получили готовые типы и аннотации, написав при этом только сами запросы и отметив их декоратором.

Теперь мы можем импортировать наши аннотированные функции-обертки, а также сгенерированные типы данных и свободно использовать их в своем проекте.

Итог

В результате мы используем SQL запросы в форме обычных вызовов функций, а аргументы в эти запросы вставляем так, как каждому интуитивно хотелось сделать в самом начале знакомства с любой SQL базой: при помощи f-string.

• Никаких SQL билдеров, никаких ORM, никаких моделей - просто запросы и их результаты в любой удобной для проекта форме.

• Когда появятся новые запросы, изменятся / удалятся старые или изменится схема базы данных, мы просто перезапустим querky_gen.py.
  
  Через секунду получим актуальные типы, не переписывая ни единой строчки кода.

• Нам больше не нужно следить за тем, используем ли мы правильные названия колонок, существует ли колонка в кортеже, какой тип имеет возвращенная колонка и т.д.
  
  За нас эту работу выполняет линтер.

• Если мы написали неправильный SQL запрос, то querky_gen.py завершится с ошибкой и указанием, в каком именно запросе она произошла.
  
  Этот приятный сайд-эффект происходит, благодаря тому что для инференции типов запросы отсылаются на ваш RDBMS, где происходит PREPARE запроса. Таким образом, база не может обработать неверные запросы и выбрасывает ошибку.
  
  Это позволяет всегда быть на 100% уверенным, что каждый запрос в целом проекте обязательно запустится в рантайме.

• Также благодаря тому, что мы работаем на живой базе с нативным SQL, в запросах можно использовать любые доступные функции, хранимые процедуры - вообще все, на что ваш RDBMS способен - и все равно весь type inference будет работать как часы.

Внимательный читатель может отметить, что не все так просто, ведь бывают NULLABLE колонки, Optional аргументы, неточности типов данных внутри базы, случается необходимость использования одного и того же возвращаемого типа данных в нескольких разных запросах, использования значений по-умолчанию в аргументах, переиспользования запросов.

Я специально решил не перегружать эту статью деталями, а решения всех проблем сверху описаны в документации. Проблемы кликабельны :)

Отмечу также, что библиотека неинвазивна: можно использовать для работы с уже готовым проектом, если тот написан на поддерживаемом querky драйвере (только asyncpg на данный момент).

Похожие проекты

• sqlx - для rust. Не генерирует типы, но тоже проверяет выполняемость запросов во время компиляции и поддерживает распаковку в нужные structы.

• sqlc - для go. Вы пишете sql типы (таблицы) и запросы к ним в отдельных .sql файлах, а библиотека генерирует go код для них. На мой взгляд могут возникнуть проблемы из-за того что информация о типах берется из .sql файлов, а не из базы напрямую. Но в этом вопросе лучше меня не слушать, а спросить тех, кто имел опыт с этим инструментом.

• pugsql - для python. Вдохновитель sqlc для go. Принцип похож но типы писать не нужно, потому как возвращаются dictы. Нет типизации.

• hugsql - для closure. Вдохновитель sqlc для go. Вроде pugsql.

Проект на querky

Как и обещал, мой проект, в котором эта библиотека уже трудится. К сожалению, не могу похвастаться большим онлайном, но приглашаю всех оценить игру:

https://t.me/cardusbot

Игра создается в любом чате, где есть бот. Игра в WebApp, т. е. в браузере. Геймплей много кому знаком: карточная игра "Дурак", по желанию можно включить множество различных модификаций и дополнений к игре.

Весь движок игры сделан целиком на SQL запросах, каждый из которых проходит через querky. Всего в приложении на данный момент 222 уникальных запроса, из которых 129 отвечают за игровой процесс.

Хочу также отметить интересную фишку, которая еще более упростила мне жизнь при разработке этого проекта.

Я все-таки использовал ORM, но в качестве снапшота состояния базы. И использовал небольшой хак, чтобы делать так:

@qrk.query('Referral', shape='many', dict=True)
def update_set_referrals(game_id, host_account_id):
    return f'''
        WITH 
        locked_account_id AS (
            SELECT 
                {account.id}
            
            FROM
                {account}
            INNER JOIN
                {player}
            
            ON
                {player.account_id} = {account.id} 
            
            WHERE
                {player.game_id} = {+game_id} 
                AND {account.new}
                AND {account.id} <> {+host_account_id}
        
            ORDER BY
                {account.id}
        
            FOR UPDATE
        )
        UPDATE
            {account}
        SET
            {~account.new} = FALSE,
            {~account.referred_by} = {+host_account_id}
        FROM
            {player}
        WHERE
            {account.id} = ANY(SELECT id FROM locked_account_id)
        RETURNING
            {account.id},
            {account.name},
            {account.tg_id}
        '''

То есть видно, что вместо строковых названий столбцов и таблиц я использую какие-то объекты.

Вот код для создания таких объектов:

from __future__ import annotations

import typing


T = typing.TypeVar('T')


class Column:
    def __init__(self, t: Table, name: str):
        self.table = t
        self.name = name

    def _str(self) -> str:
        if self.table.noname__:
            return self.name
        else:
            return f"{self.table.name__}.{self.name}"

    def __invert__(self):
        return self.name

    def __str__(self):
        return self._str()

    def __repr__(self):
        return self._str()


class Table:
    def __init__(self, name: str, noname: bool):
        self.name__ = name
        self.noname__ = noname

    def __getattr__(self, item: str) -> Column:
        return Column(self, item)

    def __str__(self):
        return self.name__

    def __repr__(self):
        return self.name__


def get_database_name(t) -> str:
    return t.__tablename__


def table(t: T, name: str | None = None) -> T:
    return Table(name or get_database_name(t), noname=False)

И тогда мы используем функцию table таким образом, подавая на вход обычные sqlalchemy модели:

from mvc.models.sql.alchemy import (
    Player,
    Account
)

player = table(Player)
account = table(Account)

В этом месте мы обманываем пайчармовский линтер. Он думает, что player и account - это sqlalchemy модели из-за того, что функция table использует T на входе и как будто бы возвращает T на выходе.

Таким образом мы еще более упростили свою жизнь:

• линтер будет подсвечивать доступ к неверным полям внутри каждого такого запроса,

• а также будет автокомплит ко всем полям в таблице - очепяток не будет.

Я не включил этого в библиотеку, потому что это скорее хак, и вряд ли очень надежный и переносимый на все IDE метод. Но получилось очень удобно, поэтому не грех поделиться.

Контакты

• Почта: verolomnyy@gmail.com

• TG: datscrazy

• github: racinette

• SO: winwin