Когда хочется строить бэкенд по DDD и CQRS на Python, а не «просто REST», приходится самому раскладывать роуты, команды, запросы и события. Я сделал фреймворк Urich — в нём один объект описывает ограниченный контекст, а маршруты и OpenAPI получаются из этого описания. Расскажу, зачем это нужно и как выглядит в коде.
В чём проблема
FastAPI и Starlette отлично закрывают задачу «принять HTTP, провалидировать, ответить». Но они не говорят, как организовать домен: где команды, где запросы, как связать агрегаты с репозиториями и кто подписывается на события. Всё это ты проектируешь сам и вручную вешаешь на роуты.
В итоге:
-
Роуты размазаны по роутерам и декораторам.
-
Конвенции (например,
POST /orders/commands/create_order) приходится соблюдать самому. -
OpenAPI для команд и запросов — опять ручная работа или копипаст.
-
Event bus, discovery, RPC — каждый раз своя обвязка.
Хочется один стиль: «один объект = один ограниченный контекст», и из него уже вытекают маршруты, доки и DI.
Что такое Urich
Urich — это асинхронный фреймворк поверх Starlette, заточенный под DDD и CQRS. Идея простая:
-
Ты описываешь ограниченный контекст одним объектом: агрегат, репозиторий, команды, запросы, обработчики доменных событий.
-
Регистрируешь его в приложении:
app.register(orders_module). -
Фреймворк сам добавляет HTTP-маршруты, OpenAPI/Swagger и прокидывает зависимости (репозитории, EventBus) в обработчики.
Никаких декораторов над каждым эндпоинтом — структура задаётся описанием модуля.
Как это выглядит в коде
Один модуль = один контекст
Файл orders/module.py:
from urich.ddd import DomainModule from .domain import Order, OrderCreated from .application import CreateOrder, CreateOrderHandler, GetOrder, GetOrderHandler from .infrastructure import IOrderRepository, OrderRepositoryImpl def on_order_created(event: OrderCreated) -> None: ... orders_module = ( DomainModule("orders") .aggregate(Order) .repository(IOrderRepository, OrderRepositoryImpl) .command(CreateOrder, CreateOrderHandler) .query(GetOrder, GetOrderHandler) .on_event(OrderCreated, on_order_created) )
Один объект — полное описание контекста «заказы». Дальше в main.py:
from urich import Application from orders.module import orders_module app = Application() app.register(orders_module) app.openapi(title="My API", version="0.1.0")
Запуск: uvicorn main:app --reload. Появятся маршруты:
-
POST /orders/commands/create_order -
GET /orders/queries/get_order(и POST для того же запроса)
И страница /docs с Swagger, где схемы запросов построены из твоих dataclass команд и запросов.
Домен и приложение
Агрегат и события — в domain.py:
from urich.domain import AggregateRoot, DomainEvent from dataclasses import dataclass @dataclass class OrderCreated(DomainEvent): order_id: str customer_id: str total_cents: int class Order(AggregateRoot): def __init__(self, id: str, customer_id: str, total_cents: int): super().__init__(id=id) self.customer_id = customer_id self.total_cents = total_cents self.raise_event(OrderCreated(order_id=id, customer_id=customer_id, total_cents=total_cents))
Обработчики команд и запросов — в application.py. Репозиторий и EventBus приходят через конструктор (DI):
from urich.ddd import Command, Query from urich.domain import EventBus @dataclass class CreateOrder(Command): order_id: str customer_id: str total_cents: int class CreateOrderHandler: def __init__(self, order_repository: IOrderRepository, event_bus: EventBus): self._repo = order_repository self._event_bus = event_bus async def __call__(self, cmd: CreateOrder) -> str: order = Order(id=cmd.order_id, customer_id=cmd.customer_id, total_cents=cmd.total_cents) await self._repo.add(order) for event in order.collect_pending_events(): await self._event_bus.publish(event) return order.id
Репозиторий объявляешь интерфейс + реализацию (in-memory для прототипа, потом БД) — фреймворк регистрирует их в контейнере и подставляет в обработчики.
Чем Urich отличается от FastAPI
FastAPI отвечает на вопрос: «как принять HTTP и отдать ответ». Роуты и модели — ты проектируешь сам. DDD, CQRS, границы контекстов — тоже на тебе.
Urich отвечает на вопрос: «как разложить сервис по ограниченным контекстам и CQRS». Ты описываешь контекст (агрегат, команды, запросы, события) — маршруты и OpenAPI получаются из этого описания. Один и тот же стиль используется для домена, шины событий, discovery и RPC.
То есть это не замена FastAPI «по роутингу», а слой выше: структура приложения по DDD, а под капотом по-прежнему Starlette (и при желании можно комбинировать с другими ASGI-частями).
Что ещё входит
-
EventBusModule — шина событий: in-memory для прототипов или свой адаптер (Redis, Kafka и т.д.).
-
OutboxModule — контракт для transactional outbox (хранение и публикация событий).
-
DiscoveryModule — разрешение имён сервисов в URL (статическая карта или свой адаптер, например Consul).
-
RpcModule — приём и вызов RPC между сервисами; опционально транспорт HTTP+JSON из коробки.
В ядре только протоколы; реализации подключаешь сам или берёшь готовые. Зависимости прокидываются через контейнер.
Есть CLI для скелета приложения и контекстов:
pip install "urich[cli]" urich create-app myapp cd myapp urich add-context orders --dir . urich add-aggregate orders Order --dir .
Для кого это
Тем, кто:
-
проектирует микросервисы или бэкенд по DDD/CQRS;
-
хочет один стиль для домена, событий и RPC без тяжёлого фреймворка;
-
готов сам выбирать персистентность и транспорт, но хочет конвенции по командам/запросам и маршрутам.
Не подходит как «замена всему»: нет ORM, админки, фронта — только организация бэкенда вокруг ограниченных контекстов.
Ссылки
-
Репозиторий: github.com/KashN9sh/urich
-
Документация: kashn9sh.github.io/urich
-
PyPI:
pip install urich
Пример полной сборки (ecommerce с заказами, event bus, discovery, RPC) лежит в репозитории в examples/ecommerce.
Буду рад обратной связи: чего не хватает для реальных проектов, какие адаптеры или паттерны было бы полезно добавить. Контрибуции (issues, PR, идеи) приветствуются.
Автор: KashN9sh
