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

Ссылки

Пример полной сборки (ecommerce с заказами, event bus, discovery, RPC) лежит в репозитории в examples/ecommerce.

Буду рад обратной связи: чего не хватает для реальных проектов, какие адаптеры или паттерны было бы полезно добавить. Контрибуции (issues, PR, идеи) приветствуются.