Всем привет! Меня зовут Юрчик Олег, я старший Python разработчик в Cloud.ru и моя основная специализация — backend-разработка. Не секрет, что работа с базами данных для backend-разработчика — это база. Каждый из нас должен знать, что такое БД, понимать, как они работают и уметь с ними работать. Когда вы пишете курсовую в университете, делаете тестовое задание, мучаетесь ночами с пет-проектом или работаете над High-Load AI Six-Seven B2B SaaS — везде вас поджидают базы данных.

Я программирую уже 16 лет, как минимум 10 из них — это ежедневная работа с СУБД. И я до сих пор плохо знаю SQL. За все время я так и не выучил расшифровку аббревиатуры ACID, иногда путаюсь в уровнях изоляции, а написать простой запрос с использованием различных JOIN или HAVING могу только с Гуглом.

И дело не в том, что я плохой разработчик, хотя тоже не исключено. За все 7 лет коммерческой разработки мне ни разу не понадобилось писать сложный SQL-запрос, а больше 50% функционала популярных ORM так ни разу и не использовались мной.

ORM

Начну с истории о том, как в этой версии вселенной я дошел до момента, в котором пишу эту статью. Всю свою жизнь я использую SQLAlchemy —это единственный правильный выбор для Python-разработчика. Данный пакет встречается повсеместно и позволяет максимально гибко реализовывать разные сценарии — от самых простых задач до сложных запросов с агрегацией данных, вложенными транзакциями, подзапросами и так далее.

При этом в реальных проектах чаще всего используется только малая часть всей библиотеки. Обычно для сервиса требуется CRUD для данных и взаимосвязи между таблицами, и больше ничего. Если же есть проблемы со скоростью запросов, то в 95% случаев они решаются проставлением корректных индексов и правильным выбором между joinedload и selectinload. В итоге SQLAlchemy предлагает множество путей для реализации одной и той же задачи, ставя в тупик новичков и заставляя их изучать десятки страниц документации, чтобы понять разницу и сделать правильный выбор. В первую очередь я имею ввиду пакеты ORM и Core, но это также относится и к различным мелочам типа функций exec и execute.

Пост ORM

Со временем я понял, что для многих моих проектов чистый SQLAlchemy — это оверхед. Тогда я нашел SQLModel, библиотеку от создателя FastAPI, которая сочетает в себе гибкость SQLAlchemy и красоту моделей из Pydantic. Теперь таблицы — это не просто кастомный класс, а полноценные DTO-классы, обладающие такими замечательными свойствами, как валидация и сериализация или десериализация, при этом сохраняющие в себе свойства таблиц типа relationships и lazy loading.

Я начал использовать SQLModel везде, но, к сожалению, единственное, что она упрощает — это объявление таблиц и работу с ними. Составление и выполнение запросов остается все так же через SQLAlchemy.

В процессе использования SQLModel я очень быстро пришел к паттерну работы через репозитории. Упрощенно — это когда к каждой таблице ты разрабатываешь независимый класс-репозиторий, в котором прописываются методы для работы с записями из этой таблицы. Паттерн Repository поддерживает принцип single responsibility из SOLID, а на начальном этапе разработки это особенно полезный подход, так как позволяет в дальнейшем разделить данные на несколько БД или даже СУБД.

В какой-то момент я понял, что таскаю один и тот же кусок кода из проекта в проект. Поэтому я решил вынести его в отдельный пакет и теперь использую везде. Уверен, что вам тоже понравится.

MetaORM

Самый часто используемый подъязык в SQL — DML (Data Manipulation Language) — семейство команд, которые напрямую занимаются работой с данными внутри таблиц. DML состоит всего из четырех запросов: SELECT, INSERT, UPDATE и DELETE. Их достаточно для того, чтобы полностью управлять данными в базе. Из этого у меня родился и интерфейс для репозиториев:

class BaseRepository:
	async def get_item(...) -> BaseModel | None:
	async def get_items(...) -> AsyncGenerator[BaseModel]:
	async def create_item(...) -> BaseModel:
	async def update_items(...) -> AsyncGenerator[BaseModel]:
	async def delete_items(...) -> None:

Вот небольшой минимальный пример для работы с БД через MetaORM:

import asyncio
import uuid

from metaorm import BaseFilter, BaseRepository, BaseTable, Field, RepositorySettings

# types for repository
class UserFilter(BaseFilter):
	id: uuid.UUID
	name__ilike: str

class UserTable(BaseTable, table=True):
	__tablename__ = "users"
	
	id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
	name: str = Field(unique=True)

# repository
class UsersRepository(BaseRepository, table=UserTable, filter_=UserFilter):
	pass

# definitions
settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:")
users_repository = UsersRepository(settings=settings)

# action
async def main():
	await users_repository.create_tables()
	
	new_user = UserTable(name="Oleg")
	new_user = await users_repository.create_item(new_user)
	print("Created new user:", new_user)

asyncio.run(main())

Для создания класса репозитория надо обязательно иметь два класса:

• Класс таблицы БД в формате SQLModel

class UserTable(BaseTable, table=True):
	__tablename__ = "users"

	id: uuid.UUID = Field(primary_key=True, default_factory=uuid.uuid4)
	name: str

• Класс для фильтров — нужен для WHERE в запросах

class UserFilter(BaseFilter):
	id: uuid.UUID
	name__ilike: str

Эти классы обязательно должны указываться при объявлении репозитория:

class UsersRepository(BaseRepository, table=UserTable, filter_=UserFilter):
    ...

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

На GitHub есть недооцененная, на мой взгляд, сообществом библиотека pydantic-filters, которая позволяет декларативно указывать модели для фильтров и применять их к SQLAlchemy и FastAPI. На ее основе используется фильтрация для запросов в MetaORM. Например:

class UserFilter(BaseFilter):
	id: uuid.UUID
	name: str
	name__ilike: str

class BookFilter(BaseFilter):
	title: str
	author: str
	user: UserFilter

Фильтры могут быть вложены друг в друга (как в BookFilter вложен UserFilter), а также поддерживают разные операторы для сравнения, аналогичные Django: eq, ilike, ge, lt и т.д.

Поиск записей в БД с использованием фильтров выглядит так:

filter_ = UserFilter(name=["Oleg", "Andrey"])
pagination = OffsetPagination(limit=10)
users = [
         user async for user in users_repository.get_users(filter_=filter_, pagination=pagination)]

В этом коде мы явно запрашиваем первых 10 пользователей с именами Oleg и Andrey.

От таблиц к DTO

В примере выше мы создавали UserTable(name="Oleg") и получали обратно тот же UserTable. Это удобно для быстрого старта, но в реальном проекте хочется разделить слой хранения и слой доменных моделей. Иначе бизнес-логика начинает знать про Field, primary key и прочие детали SQL.

MetaORM позволяет отделить таблицу от DTO через generics:

from pydantic import BaseModel
from metaorm import BaseTable

class User(BaseModel):
    id: int | None = None
    name: str
    email: str

class UserTable(BaseTable[User], table=True):
    __tablename__ = "users"

    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)

    @classmethod
    def from_item(cls, item: User) -> "UserTable":
        return cls(id=item.id, name=item.name, email=item.email)

    def to_item(self) -> User:
        return User(id=self.id, name=self.name, email=self.email)

Теперь указываем dto=User при объявлении репозитория:

class UserRepository(BaseRepository, table=UserTable, filter_=UserFilter, dto=User):
    pass

И вся работа идет с чистым Pydantic-классом без ORM-примесей:

user = await repository.create_item(User(name="Alice", email="alice@example.com"))
print(type(user))  # <class 'User'>

Фреймворк сам вызовет from_item при записи и to_item при чтении. Это тот же паттерн Data Mapper, но без XML-конфигураций и метаклассов, которые делают SQLAlchemy похожим на заклинание из «Гарри Поттера».

Связи и eager loading

Когда в проекте появляется больше одной таблицы, встает вопрос: как загружать связанные данные? MetaORM не изобретает велосипед — под капотом все тот же SQLAlchemy, поэтому работают стандартные joinedload и selectinload.

from sqlalchemy.orm import joinedload
from metaorm import Field, Relationship

class AuthorTable(BaseTable, table=True):
    __tablename__ = "authors"

    id: int = Field(primary_key=True)
    name: str
    books: list["BookTable"] = Relationship(back_populates="author")

class BookTable(BaseTable, table=True):
    __tablename__ = "books"

    id: int = Field(primary_key=True)
    title: str
    author_id: int = Field(foreign_key="authors.id")
    author: AuthorTable = Relationship(back_populates="books")

При чтении передаем options:

books = [
    item
    async for item in book_repo.get_items(
        options=[joinedload(BookTable.author)],
    )
]

for book in books:
    print(f"{book.title} — {book.author.name}")

Никакой ленивой загрузки, никаких N+1. Вы явно говорите, что хотите подтянуть, и SQLAlchemy делает это одним запросом. Если нужно selectinload — просто меняете одну строчку. Для большинства задач этого достаточно, чтобы не лезть в ручной SQL.

Транзакции без боли

Каждый метод репозитория уже обернут в транзакцию автоматически. Но иногда нужно сгруппировать несколько операций в одну атомарную единицу. Для этого у репозитория есть transaction():

async with repository.transaction():
    product1 = await repository.create_item(ProductTable(name="Laptop", price=999.99))
    product2 = await repository.create_item(ProductTable(name="Mouse", price=29.99))

Все внутри async with выполняется в одной сессии. Если упадет исключение — откатится все.
При этом вложенные вызовы repository.transaction() не создают новых сессий, а переиспользуют текущую. Это удобно, когда один сервисный метод вызывает другой и каждый из них может открывать свою транзакцию:

async with repository.transaction(), repository.transaction():
    items = [item async for item in repository.get_items()]
    # Вторая transaction() просто видит, что сессия уже есть, и берёт её.

Savepoints: когда нужно откатить только часть

Иногда внутри большой транзакции нужно попробовать что-то сделать, и если не вышло — откатить только эту часть, не трогая остальное. Для этого есть nested_transaction(), который под капотом использует SQLAlchemy begin_nested() — то есть настоящий savepoint СУБД.

async with repository.transaction():
    await repository.create_item(ProductTable(name="Keyboard", price=79.99))

    try:
        async with repository.nested_transaction():
            await repository.create_item(ProductTable(name="Monitor", price=299.99))
            raise ValueError("Something went wrong")
    except ValueError:
        pass  # Monitor откатился, Keyboard остался

    items = [item async for item in repository.get_items()]
    assert len(items) == 1
    assert items[0].name == "Keyboard"

Это работает и без внешней транзакции — тогда savepoint создаётся прямо в новой сессии. Гибкость, которую в ORM обычно прячут за десятью страницами документации, здесь доступна в двух строчках.

Многорепозиторные транзакции

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

MetaORM решает это через RepositoriesContainer и contextvars:

from metaorm import RepositoriesContainer

settings = RepositorySettings(dsn="sqlite+aiosqlite:///:memory:")
container = RepositoriesContainer(settings=settings)

user_repo = container.get_repository(UserRepository)
order_repo = container.get_repository(OrderRepository)

async with container.transaction():
    user = await user_repo.create_item(UserTable(name="Alice"))
    await order_repo.create_item(OrderTable(user_id=user.id, total=100.00))
    await order_repo.create_item(OrderTable(user_id=user.id, total=250.50))

Контейнер создает AsyncSession, кладет ее в контекстную переменную, и все репозитории внутри async with автоматически используют эту сессию. Никакого пробрасывания session через пять слоев абстракции — вы просто открываете блок, и все внутри него атомарно.

То же самое работает с container.nested_transaction() для savepoint между разными репозиториями:

async with container.transaction():
    user = await user_repo.create_item(UserTable(name="Bob"))
    try:
        async with container.nested_transaction():
            await order_repo.create_item(OrderTable(user_id=user.id, total=999.99))
            raise ValueError("Rollback nested order")
    except ValueError:
        pass
    # Bob сохранился, заказ откатился

Все это строится поверх SQLAlchemy и SQLModel, и MetaORM ничего от вас не скрывает. Он дает удобную стартовую точку для рутинных задач, но не мешает лезть под капот, когда стандартных методов начинает не хватать.

Что умеет MetaORM

Вся библиотека — это пара десятков строк на декларативные фильтры, несколько генераторов для стандартных CRUD-операций и тонкий слой управления транзакциями через contextvars. Здесь нет unit-of-work, нет identity map, нет ленивых прокси и нет магического отслеживания состояния объектов. Зато есть все то, с чем backend-разработчик сталкивается каждый день:

• декларативные фильтры с операторами (__ilike, ge, in и т.д.),

• пагинация и сортировка из коробки,

• DTO mapping через generics,

• атомарные и вложенные транзакции,

• Eager loading нативными средствами SQLAlchemy,

• разделение таблицы и доменной модели.

Когда стандартных методов мало

Главное опасение при выборе «простой» библиотеки в том, что рано или поздно вы упретесь в потолок и не сможете решить нетривиальную задачу. С MetaORM такого не произойдет, потому что под капотом все тот же SQLAlchemy, а репозиторий — это обычный класс, который можно расширять как угодно.

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

from sqlalchemy import func, select


class ProductRepository(BaseRepository, table=ProductTable, filter_=ProductFilter):
    async def get_average_price(self) -> float:
        table = self.get_table_type()
        statement = select(func.avg(table.price)).select_from(table)

        async with self.transaction():
            result = await self.session.exec(statement)
            return result.scalar_one() or 0.0

self.session — это тот же AsyncSession из SQLAlchemy, а self.transaction() — ваш привычный контекстный менеджер. Вы пишете ровно тот SQLAlchemy-код, который написали бы и без MetaORM, но при этом остаетесь внутри репозитория с его соглашениями о транзакциях и конвертации результатов.

Если понадобится CTE, оконная функция, сложный JOIN с группировкой или даже сырой SQL через text() — вы делаете то же самое. Никаких заклинаний, никакого выпадения из парадигмы. MetaORM дает удобную обёртку для рутины, но не становится между вами и базой данных, когда нужна вся ее мощь.

Заключение

Я не призываю выбрасывать SQLAlchemy ORM или Django ORM из существующих проектов. Если у вас все работает — не чините. Это мощные инструменты, и в некоторых задачах они по-настоящему незаменимы.

Но за много лет разработки я убедился, что большая часть кода в типичном backend-сервисе — это одно и то же: взять запись по идентификатору, отфильтровать список по условиям, обновить пару полей, создать сущность внутри транзакции. И каждый раз я видел, как разработчики тратят часы на то, чтобы понять, почему lazy="dynamic" ведет себя не так, как ожидалось, или как правильно сконфигурировать relationship для полиморфной ассоциации, которая в проекте никогда не понадобится.

MetaORM — это мой способ упростить работу с СУБД. Хватит тащить в проект абстракции на все случаи жизни, если вы ими не пользуетесь. Хватит заставлять новичков читать пятьсот страниц документации, чтобы сделать простой CRUD. Хватит копировать один и тот же шаблонный код транзакций, фильтров и пагинации из репозитория в репозиторий.

Это не попытка заменить SQLAlchemy. Это попытка дать ему человеческий интерфейс для тех 98% задач, где нужен не швейцарский нож, а просто хороший кухонный нож. А если вам вдруг понадобится лобзик — он всегда под рукой.

Пакет, к сожалению, не доступен на PyPI, так как использует код библиотеки pydantic-filters прямо с GitHub. Но его можно установить напрямую с GitHub:

pip install git+https://github.com/OlegYurchik/metaorm

Пробуйте, форкайте, пишите issues. Буду рад любой обратной связи.