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

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

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

Ну ведь правда, согласитесь?

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

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

SQLModel — первый шаг к счастью

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

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

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

Паттерн Repository

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

Тяжёлая разница

Самый часто используемый подъязык в 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.

В примере выше мы создавали 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 похожим на заклинание из «Гарри Поттера».

Когда в проекте появляется больше одной таблицы, встает вопрос: как загружать связанные данные? 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() просто видит, что сессия уже есть, и берёт её.

Иногда внутри большой транзакции нужно попробовать что-то сделать, и если не вышло — откатить только эту часть, не трогая остальное. Для этого есть 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 ничего от вас не скрывает. Он дает удобную стартовую точку для рутинных задач, но не мешает лезть под капот, когда стандартных методов начинает не хватать.

Вся библиотека — это пара десятков строк на декларативные фильтры, несколько генераторов для стандартных 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. Буду рад любой обратной связи.