Domain Driven Design и модульный монолит в Python

Пролог

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

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

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

Почему вообще DDD и модульный монолит в Python?

По мере моей коммерческой деятельности как Python backend разработчика я часто встречался с довольно объемными, и сложными проектами, где было необходимо реализовать огромное количество функционала и сложной бизнес логики.

Если приложение получается довольно большим, то следовать самой распространенной для Python‑приложений и одновременно самой простейшей слоистой архитектуре становится довольно сложно. Границы размываются, очень часто возникают проблемы циклических импортов, появляются god‑objects, сервисная логика становится настолько захламленной, что один метод, который делает, казалось бы, небольшое изменение данных, может начать занимать больше 200 строк кода. В свое время, проводя время в поиске решений этой проблемы, я наткнулся на тему Domain Driven Design. Если вкратце — Domain Driven означает подход разработки веб приложения, который сконцентрирован вокруг бизнес‑логики и ее инвариантов.

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

Тогда почему именно модульный монолит? Ведь реализовать DDD возможно и в обычном монолите. В чем разница? Разница между обычным монолитом и модульным сводится к чёткому разделению доменных границ, которое является одним из плюсов микросервисной архитектуры, однако не усложняет инфраструктурную поддержку, деплой и разработку, потому как приложение до сих пор остается монолитом. Таким образом, архитектура модульного монолита сочетает в себе плюсы обоих этих подходов.

Реализация в коде

К сожалению, в Python нет сильных, поддерживаемых фреймворков, как, например, Spring Modulith в Java или EF/.NET в C#, которые предоставляют реализации этих паттернов нативно. В Python придётся делать все самостоятельно с нуля, что объясняет отсутствие популярности в реализации, однако не отменяет её возможность, и более того, предоставляет большую свободу выбора действий под различные ситуации. Далее я покажу, как можно реализовать на практике всё то, что объясняется в теории про эти архитектурные подходы, и начнем, пожалуй, с разбора Domain Driven Design.

Доменный уровень

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

Доменные модели — как правило представляют собой классы, которые обозначают одну сущность (агрегат), инкапсулируют присущие ей данные и бизнес‑логику. Являются ядром системы. Есть три основных способа реализации доменной модели:

1) Обычный Python класс — самый простой способ:

class Product:
  id: int
  name: str
  
  def __init__(self, id: int, name: str):
    self.id = id
    self.name = name

# Использование
product = Product(id=1, name="Lamp")

Работает без лишних импортов и сторонних библиотек, однако имеет свои недостатки, такие как необходимость реализации конструктора, метода репрезентации (если нужно), а также линтеры и типизаторы постоянно жалуются на built‑in имена полей по типу id, type и так далее.

2) Python dataclass — реализация с использованием встроенного модуля dataclasses:

from dataclasses import dataclass


@dataclass
class Product:
  id: int
  name: str

# Использование
product = Product(id=1, name="Lamp")

Из плюсов — метод конструктора, сравнения и репрезентации из коробки, имеется возможность работы с дефолтами и фабриками у полей, нет жалоб от линтеров и типизаторов. Можно определить frozen=True если нужна read‑only модель. Из минусов — импорт модуля dataclasses, в отличие от объекта обычного класса, dataclass версия по умолчанию не хешируема (нужно объявлять frozen=True), это важно, если у вас есть логика взаимодействия моделей с set и dict типами.

3) Pydantic BaseModel — реализация с использованием библиотеки pydantic:

from pydantic import BaseModel


class Product(BaseModel):
  id: int
  name: str

# Использование
product = Product(id=1, name="Lamp")

Из плюсов — огромное количество встроенного функционала, начиная от валидации, заканчивая парсингом данных, что может быть очень полезно при передаче данных из одного слоя в другой. Однако домен начинает зависеть от сторонней библиотеки, что, по моему мнению, не слишком сильный редфлаг, потому как равных по развитию, поддержке и популярности pydantic в Python нет, и заменять pydantic в Python проекте на что‑то иное практически не имеет смысла. Объект обычно так же не хешируем, пока не объявлен model_config = ConfigDict(frozen=True).

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

# Переопределяем метод конструктора
class Product(BaseModel):
  id: UUID
  name: str
  status: ProductStatus
  def __init__(self, name: str):
    super().__init__(
      id=uuid4(),
      name=name,
      status=ProductStatus.ACTIVE,
    )

    
# Или пишем свой метод если хотим напрямую обозначить действие
class User(BaseModel):
  id: UUID
  name: str
  
  @classmethod
  def register(cls, name: str) -> "User":
    return cls(
      id=uuid4(),
      name=name
    )

Объекты значений (value objects) — служат как определения одного/группы полей доменной модели, инскапсулируя собственную валидацию внутри себя, тем самым очень сильно помогают разгружать логику валидации внутри доменной модели если таковой много. Может представлять из себя любой тип данных начиная с обычного type alias заканчивая отдельным классом который группирует в себе несколько полей.

from uuid import UUID
from enum import StrEnum
from dataclasses import dataclass


type EntityId = UUID


class Currency(StrEnum):
  UZS = "uzs"
  USD = "usd"

  
# Можно использовать обычный класс или Pydantic, как в примере с моделями
@dataclass
class Money:
  currency: Currency
  amount: int
  
  def __init__(self, currency: Currency, amount: int):
    if amount <= 0:
      raise ValueError("Money amount should be greater than zero")
      
    self.currency = currency
    self.amount = amount
    
  def convert(self, to: Currency):
    if self.currency != to:
      if to == Currency.USD:
        self.amount = int(self.amount / 12000)
      else:
      self.amount = self.amount * 12000
      self.currency = to

Исключения — обычно доменные исключения наследуют и переопределяют базовое кастомное исключение, реализованное на общем уровне монолита.

# Общий уровень
class DomainError(Exception):
  status: int = 500
  code: str = "internal_server_error"
  message: str
  
    def __init__(self, message: str):
      self.message = message

      
# Переопределение в домене
class OrderNotFound(DomainError):
  status = 404
  code = "order_not_found"

  
class InvalidOrderStatus(DomainError):
  status = 400
  code = "invalid_order_status"

  
# Использование
def cancel_order(repo: OrderRepo, order_id: int):
  order = repo.get_by_id(order_id)
  if order is None:
    raise OrderNotFound(f"Order {order_id} not found")
    
  order.cancel()

  
class Order:
  id: int
  status: OrderStatus
  
  def cancel(self):
  if self.status != OrderStatus.ACTIVE:
    raise InvalidOrderStatus(
      f"Invalid order {self.id} status {self.status}, expected status
      {OrderStatus.ACTIVE} to cancel"
    )
  self.status = OrderStatus.CANCELLED

В продакшн разработке рекомендуется отдавать клиенту ответ, содержащий как минимум три вещи — статус, код ошибки и сообщение, подробно описывающее проблему. Я не вижу ничего зазорного в том чтобы определять значения статуса и кода ошибки на уровне домена, иначе придется создавать сложную и очень большую таблицу маппинга доменных исключений в HTTP ответы, что я считаю излишней сложностью. Такой компромисс дает нам гибкость: прокидывая исключение наверх таким образом, мы передаем ответственность за хендлинг ошибки следующему слою, если предусмотрены альтернативные способы обработки логики на случай ошибки, или формирование ответа для клиента, чем обычно обычно занимаются exception handlers (если мы говорим про FastAPI) или самописная middleware.

События — представляют собой классы (обычные, Pydantic или датаклассы), инкапсулирующие плоские данные (без вложенных классов, без отдельных методов кроме конструктора), нужные для уведомления о возникшем изменении в одной части системы для других частей, без их прямого вызова. Могут генерироваться как внутри доменной модели после определенных действий, так и на аппликационном уровне, например если производятся batch действия над несколькими сущностями одновременно, или нужно уведомить систему о каком либо действии, не связанном ни с одной сущностью.

from pydantic import BaseModel
from dataclasses import dataclass, field


# Родительский класс на общем уровне
class DomainEvent(BaseModel):
  pass

# Класс доменной модели, содержащий хранилище событий
@dataclass
class DomainModel:
  _events: list[DomainEvent] = field(default_factory=list, init=False)

# Наследование на уровне домена
class OrderCancelled(DomainEvent):
  order_id: int


# Использование
@dataclass
class Order(DomainModel):
  id: int
  status: OrderStatus
  def cancel(self):
    self.status = OrderStatus.CANCELLED
    self._events.append(OrderCancelled(order_id=self.id))

Инфраструктурный уровень

На этом уровне реализуется код который отвечает за взаимодействие с внешними узлами системы, такими как бд, внешний API, кеш и так далее. Разберем основные моменты:

Взаимодействие с базами данных — согласно DDD домен должен быть отделен и независим от инфраструктурных данных, поэтому классы и типы, предоставляемые библиотеками для работы с базами данных, не следует использовать непосредственно в качестве доменных моделей. Потому как сегодня вы можете использовать SQLAlchemy и Postgres в качестве хранилища данных, а завтра переедете на Motor от MongoDB и вам придется изменять доменный слой конкретно под используемую бд, что противоречит самой сути Domain Driven подхода. Вместо этого стоит определять их отдельно, и маппить друг на друга в момент получения и сохранения данных. Вы заплатите цену небольшого оверхеда за чистоту кода и взаимозаменяемость компонентов системы.

Разберем на примере SQLAlchemy (для nosql или иных sql‑oriented библиотек принципы те‑же), по сути все сводится всего к двум подходам реализации маппинга доменных моделей на типы данных бд и наоборот:

1) Реализация отдельного маппера: Это может быть отдельный набор методов, класс, любая структура которая содержит в себе реализацию маппинга, представляющая собой связующее звено между доменной моделью и базой данных. К примеру есть ORM модель вида:

class ProductORM(Base):
  __tablename__ = "products"
  
  id: Mapped[int] = mapped_column(Integer, primary_key=True)
  name: Mapped[str] = mapped_column(String)

Мы хотим маппить ее в нашу доменную модель на чистом Python и наоборот:

@dataclass
class Product:
  id: int
  name: str
  
  def update_name(self, name: str):
    self.name = name

Реализуем класс маппера:

class ProductMapper:
  @staticmethod
  def map_domain_to_orm(domain: Product) -> ProductORM:
    return ProductORM(
      id=domain.id,
      name=domain.name
    )
    
  @staticmethod
  def map_orm_to_domain(orm: ProductORM) -> Product:
    return Product(
      id=orm.id,
      name=orm.name
    )
    
  @staticmethod
  def map_domain_to_values(domain: Product) -> dict:
    return {
      "name": domain.name
    }

Последний метод нужен для обновления записи через batch update с использованием values, так как в доменном слое мы взаимодействуем именно с Python классом а не с ORM напрямую.

class ProductRepo:
  def __init__(self, session: AsyncSession, mapper: ProductMapper):
    self.session = session
    self.mapper = mapper
    
  async def find_by_id(self, product_id: int) -> Product | None:
    orm = await self.session.get(ProductORM, product_id)
    if orm is None:
      return None
    
    return self.mapper.map_orm_to_domain(orm)
  
  async def save(self, product: Product) -> None:
    orm = self.mapper.map_domain_to_orm(product)
    self.session.add(orm)
    
  async def update(self, product: Product) -> None:
    values = self.mapper.map_domain_to_values(product)
    await self.session.execute(
      update(ProductORM)
      .where(ProductORM.id == product.id)
      .values(**values)
    )

2) Реализация маппинга через использование возможностей библиотеки: Некоторые библиотеки для взаимодействия с базами данных предоставляют готовые способы маппинга полученных данных из бд в обычные Python классы, в том числе и SQLAlchemy, там это возможно реализовать через императивный (классический) маппинг:

product_table = Table(
  "products",
  metadata,
  Column("id", Integer, primary_key=True),
  Column("name", String, nullable=False)
)

# Класс Product это тот же самый dataclass что я показывал ранее
mapper_registry.map_imperatively(
  Product,
  product_table
)

Таким образом мы можем напрямую взаимодействовать с нашим Python объектом как с ORM, не внося никаких инфраструктурных деталей SQLAlchemy в домен.

Подробнее об императивном маппинге SQLAlchemy или возможностях используемой вами библиотеки для взаимодействия с бд читайте в соответствующей документации.

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

from abc import ABC, abstractmethod


class IProductRepo(ABC):
  @abstractmethod
  async def find_by_id(self, product_id: int) -> Product | None:
    raise NotImplementedError()
    
  @abstractmethod
  async def save(self, product: Product) -> None:
    raise NotImplementedError()

    
class ProductRepo(IProductRepo):
  def __init__(self, session: AsyncSession):
    self.session = session
    
  async def find_by_id(self, product_id: int) -> Product | None:
    return await self.session.get(Product, product_id)

  async def save(self, product: Product) -> None:
    self.session.add(product)

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

def get_repo(session: AsyncSession) -> IProductRepo:
  return ProductRepo(session=session)

async def service():
  async with session_factory() as session:
    repo = get_repo(session)
    await repo.session.commit() # Unresolved attribute reference 'session' for class 'IProductRepo'

Аппликационный уровень

На этом уровне происходит оркестрация взаимодействия доменной логики с инфраструктурой. Основные компоненты:

DataTransferObjects / Схемы / Response models — Эти компоненты отвечают за инкапсуляцию и транспортировку данных внутрь и вовне аппликационного уровня. По сути все три перечисленных термина это синонимы, выбор названия которых варьируется от проекта к проекту. Представляют собой классы, содержащие какие либо данные, нужные для обработки или уже обработанные и готовые к транспортировке клиенту.

# Пример DTO для создания продукта с использованием dataclass
@dataclass(frozen=True)
class CreateProductDTO:
  name: str
  price: int

# Пример Pydantic схемы для выдачи данных клиенту
class CategoryRef(BaseModel):
  name: str
  priority: int


class ProductOut(BaseModel):
  id: int
  name: str
  categories: list[CategoryRef]

На практике у меня в основном все сводилось к использованию Pydantic, как для данных ввода, вывода так и для транспортировки значений между слоями приложения. При этом я не считаю необходимым реализовывать отдельные DTO схемы для каждого слоя, по той причине которую я уже упоминал выше (Pydantic по сути монополист на рынке Python). Такой подход сводит к минимуму бойлерплейт с маппингом данных на каждый слой и снижает когнитивную нагрузку.

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

1) Отдельные методы в одном Python‑модуле:

async def create_product(uow: UnitOfWork, repo: ProductRepo, data: CreateProductData) -> ProductOut:
  async with uow:
    product = Product.create(name=data.name, category=data.category)
    await repo.save(product)
    
  return ProductOut.model_validate(product, from_attributes=True)
  
async def cancel_order(uow: UnitOfWork, repo: OrderRepo, data: CancelOrderData) -> None:
...

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

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

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

2) Единый сервисный класс:

class ShopService:
   def __init__(self, uow: UnitOfWork, product_repo: ProductRepo, order_repo: OrderRepo):
     self.uow = uow
     self.product_repo = product_repo
     self.order_repo = order_repo
    
   async def create_product(self, data: CreateProductData) -> ProductOut:
     async with self.uow:
       product = Product.create(name=data.name, category=data.category)
       await self.product_repo.save(product)
       return ProductOut.model_validate(product, from_attributes=True)
    
   async def cancel_order(self, data: CancelOrderData) -> None:
     ...

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

Однако это будет являться минусом если класс обрастает множеством разношерстных зависимостей, из которых каждый метод будет использовать всего пару штук во время runtime. Конечно можно разделить сервисы по принципу SRP:

class ProductService:
  def __init__(self, uow: UnitOfWork, repo: ProductRepo):
    self.uow = uow
    self.repo = repo
  
  async def create(self, data: CreateProductData) -> ProductOut:
    async with self.uow:
      product = Product.create(name=data.name, category=data.category)
      await self.repo.save(product)
  
    return ProductOut.model_validate(product, from_attributes=True)
  
  
  class OrderService:
    def __init__(self, uow: UnitOfWork, repo: OrderRepo):
      self.uow = uow
      self.repo = repo

    async def cancel(self, data: CancelOrderData) -> None:
      ...

Но даже тут могут возникнуть проблемы: если для обработки одного агрегата нужны будут разные инфраструктурные зависимости, которые могут не использоваться конкретным хендлером, но открывать свои контексты (если вы делаете это в DI), брать соединения из пула и так далее

3) (Похоже на то что реализуется в Java или C#) Отдельный класс на каждое действие:

class CreateProductHandler:
  def __init__(self, uow: UnitOfWork, repo: ProductRepo):
    self.uow = uow
    self.repo = repo
  
  async def handle(self, data: CreateProductData) -> ProductOut:
    async with self.uow:
      product = Product.create(name=data.name, category=data.category)
      await self.repo.save(product)
  
    return ProductOut.model_validate(product, from_attributes=True)

  
class CancelOrderHandler:
  def __init__(self, uow: UnitOfWork, repo: OrderRepo):
    self.uow = uow
    self.repo = repo
  
  async def handle(self, data: CancelOrderData) -> None:
    ...

Этот подход сочетает в себе плюсы обоих предыдущих, однако если таких хендлеров становится много, то это может привести к оверхеду из за постоянно встречающихся методов конструктора, а также разрастанием фабрик в точке сборки приложения (там где происходит wiring и формируется DI контейнер).

Точка входа (презентационный уровень)

Надолго останавливаться на этом уровне не буду, это входной уровень, на котором реализуется все что связано с обращениями внешних клиентов к приложению. API роутеры, веб сокеты, потребление событий, даже запуск крон тасок и джобов. Любой вызов основной логики начинается здесь. Основной момент который можно обсудить на этом уровне это инъекция зависимостей, которая может быть реализована через нативные механизмы Web фреймворка или же предоставленного DI фреймворком контейнера зависимостей.

Также возможно реализовать свой собственный DI контейнер используя встроенные возможности языка Python. Но зачастую на практике удобнее использовать уже готовый, потому как с ростом приложения возникает ряд сложностей, появляется нужна в различных уровнях контекста (синглтоны, per‑request), приходится реализовывать логику открытия и закрытия и так далее

Теперь, поговорив об основных концепциях DDD, можно обсудить, как же все таки реализовать модульный монолит?

Архитектура проекта

Основываясь на изученной информации я пришел к одному простому выводу — что модуль, обычно, это доменная область (или же микросервис), а модульный монолит это хранилище таких модулей в одной единице деплоя. Один модуль содержит в себе сущности, присущие только его области определения. Как например модуль Shop содержит в себе сущности продуктов, заказов, инфраструктуру магазина, может полноценно работать только с ними и не может напрямую импортировать, создавать, изменять сущности из области другого модуля или реализовывать не относящуюся к магазину инфраструктуру. Один модуль может лишь запрашивать данные или изменения из другого модуля, используя прямую коммуникацию через публичную точку входа, или же уведомлять другие модули о своих внутренних изменениях (такой‑же подход как и в микросервисах).

Это дает системе расширяемость, разделяет большие задачи на множество маленьких и легко изменяемых, что может дать значительный прирост в скорости разработки за счет снижения когнитивной нагрузки (вспомните кучу взаимных импортов и хендлеры на 200 строк каждый). Это и есть разделение ответственности.

Я бы выделил два основных слоя приложения:

Межмодульная коммуникация

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

1) Публичный шлюз — В микросервисном подходе таковым является внутрисистемный API, предоставляемый одним сервисом специально для внутренних компонентов системы. Но так как мы пишем монолит, то это позволяет нам обращаться к такому шлюзу без внешних сетевых запросов. Существует два известных мне подхода к реализации, один из них заключается в том что шлюз только лишь вызывает предоставляемые аппликационным слоем хендлеры/сервисы, что позволяет переиспользовать их для других клиентов:

# src/modules/shop/presentation/public.py
# Так как публичный шлюз это точка входа - реализуем на презентационном уровне
from src.modules.shop.application import handlers, commands
from src.modules.shop.domain.infrastructure.repo import ProductRepo
from src.core.sqlalchemy import UnitOfWork


# Принципы те же что и с реализацией сервисного слоя
# Можете применять SRP или наоборот объединять логику в зависимости от ситуации
class ShopPublicGateway:
  # Прокидываем зависимости на уровне инъекции зависимостей
  def __init__(self, uow: UnitOfWork, product_repo: ProductRepo):
    self.uow = uow
    self.product_repo = product_repo
  
# Рекомендуется принимать и отдавать примитивы, чтобы разорвать зависимость между модулями
# Если не получается, реализуйте публичные DTO и ACL на стороне вызывающего
  async def check_product_reliability(self, product_id: int) -> bool:
    command = commands.CheckProductReliability(product_id=product_id)
    return await handlers.check_product_reliability(uow=self.uow, repo=self.product_repo, command=command)

  
# src/modules/storage/application/handlers.py
# Прокидываем публичный gateway магазина на уровне инъекции зависимостей
async def replenish_stock_item(uow: UnitOfWork, repo: StockRepo, shop_gateway: ShopPublicGateway, command: ReplienishStockItem) -> None:
  async with uow:
    item = await repo.find_by_id(command.item_id)
    if item is None:
      raise StockItemNotFound(...)
      
    # Делаем прямой in-memory вызов
    reliable = await shop_gateway.check_product_reliability(product_id=item.product_id)
    if reliable:
      item.replenish(command.quantity)

Можно скрыть реализацию публичного шлюза за интерфейсом, также как мы делали ранее для инфраструктуры:

# src/modules/shop/presentation/public.py
from abc import ABC, abstractmethod


class IShopPublicGateway(ABC):
  @abstractmethod
  async def check_product_reliability(self, product_id: int) -> bool:
    raise NotImplementedError()


class ShopPublicGateway(IShopPublicGateway):
  ...

  
# src/modules/storage/application/handlers.py
async def replenish_stock_item(uow: UnitOfWork, repo: StockRepo, shop_gateway: IShopPublicGateway, command: ReplienishStockItem) -> None:
  ...

Второй подход подразумевает реализацию отдельных методов, заточенных специально под публичный контракт, прямо внутри шлюза:

from src.modules.shop.domain.models import Product
from src.modules.shop.domain.infrastructure.repo import ProductRepo
from src.core.sqlalchemy import UnitOfWork


class ShopPublicGateway:
  def __init__(self, uow: UnitOfWork, product_repo: ProductRepo):
    self.uow = uow
    self.product_repo = product_repo

  # Больше не вызываем аппликационный слой, а реализуем логику прямо здесь
  async def check_product_reliability(self, product_id: int) -> bool:
    async with self.uow:
      product = self.product_repo.find_by_id(product_id)
      if product is None:
        return False

    return product.is_reliable

Суть та же самая как если бы вы делали API роутеры. И да, вам все же придется делать импорты публичного шлюза между модулями, однако это куда лучше, чем тянуть все сервисы, инфраструктуру и модели напрямую, так как в будущем (особенно если вы реализовали интерфейс), при выносе отдельных или всех модулей в микросервисы вам не составит труда заменить публичный gateway модуля на HTTP клиент, вместо того чтобы переписывать все приложение.

2) Генерация и обработка событий — Разница между прямым вызовом логики другого компонента системы, и событийно‑ориентированном подходе в том, что те компоненты, которые хотят знать о произошедшем событии, сами подписываются и слушают эти события (Путем реализации шины сообщений, которая сама будет вызывать логику зарегистрированных подписчиков, или использованием стороннего брокера сообщений по типу Kafka или RabbitMQ). Это позволяет избавиться от типичной для спагетти‑кода логики, когда в одном методе помимо реализации основного действия присутствует реализация побочных эффектов. Это способствует разделению ответственности между компонентами системы, что в свою очередь открывает возможность для большего масштабирования. Тут я вынужден показать на картинках так как кода довольно много, реализацию через in‑memory message bus и внешний брокер сообщений вы сможете глянуть в моем GitHub репозитории

Можно разделить event‑driven архитектуру на две основные реализации:

Тут событие обрабатывается полностью в оперативной памяти приложения, без использования стороннего брокера сообщений, отличие от которого у шины в том, что она сама вызывает все зарегистрированные на событие хендлеры. Вызов может происходить как последовательно так и параллельно (в async режиме — конкурентно).

Также существуют реализации шины совместно с thread‑safe или межпроцессной очередью, где применяется стратегия reader‑writer на уровне OS. Но это больше походит на брокер, поэтому порой проще применить уже готовый по типу Kafka, RabbitMQ вместо того чтобы реализовывать самописный с нуля.

Используется в приложении, где нужно реализовать стратегию reader‑writer (pub/sub), где паблишером выступает инициатор события, а подписчиками — его потребители. Брокером сообщений могут выступать такие утилиты как RabbitMQ, Kafka, Redis и множество других на ваш выбор. В качестве фремворка потребителя на Python можно использовать FastStream, один из самых популярных и поддерживаемых на рынке на момент написания статьи. Обычно это используется в микросервисной архитектуре, и реализуется тяжелее чем событийная шина.

Если попытаться реализовать этот подход в монолите то можно столкнуться с большой трудностью (по крайней мере мне было трудно когда я только начинал изучать этот вопрос) — придется делать общую кодовую базу для Web запросов, и для потребителя событий, чтобы избежать дублирования всей логики приложения.

Благо разделение слоев и инъекция зависимостей помогает решить эту проблему. Если вы будете реализовывать данный подход, то вам придется изолировать все слои вашей кодовой базы от внутренностей того или иного фреймворка, и оставить их только в точке входа. FastAPI и FastStream имеют схожий DI механизм, что позволяет нам обойтись нативным функционалом языка (фабрики), однако если вы решите использовать иной стек, то вам скорее всего придется воспользоваться DI фреймворками.

Но есть еще одна, не менее важная проблема в событийно ориентированной архитектуре — атомарность при отправке событий. Прежде я показывал прямой вызов и отправку/реализацию события, но что если приложение упадет во время обработки события? Наша система останется не консистентной.

Тут на помощь приходит паттерн Outbox! (черный ящик)

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

Так и обработать его самостоятельно:

Таким образом запрос все операции производятся атомарно, система становится устойчива к падениям.

Эпилог

Конечно данная архитектура подойдет не всем проектам, простой лендинговой страничке не нужно разделение на модули или применение DDD (а то и вовсе сам бекенд), однако если вы сталкивались с проблемами при масштабируемости в больших Python проектах, то возможно среди перечисленных мною подходов вы нашли что то полезное для себя.

Полезные материалы:

Robert C. Martin — Clean Architecture: A Craftsman's Guide to Software Structure and Design: Книга описывающее применения подхода чистой архитектуры, полезно изучить для понимания определения границ и зависимостей.

Синяя книга: Eric Evans — Domain‑Driven Design: Tackling Complexity in the Heart of Software: Оригинальная работа, полностью описывающая основные концепции DDD.

Красная книга: Vaughn Vernon — Implementing Domain‑Driven Design: Продолжает все те идеи что описаны в синей книге с демонстрацией практического применения в реальных системах.

Cosmic Python — работа, подробно описывающая принципы построения сложных систем на Python, в том числе с использованием DDD.

Работы.NET разработчиков, в которых описываются принципы построения модульного монолита:
https://www.kamilgrzybek.com/blog/posts/modular‑monolith‑primer/ https://milanjovanovic.tech/blog/what‑is‑a‑modular‑monolith/

Блог Python разработчика в котором автор подробно описывает и реализация некоторых принципов, которые я упоминал (Entities, Outbox, CommandBus, Handlers, DomainModels):
https://blog.szymonmiks.pl/

Еще примеры реализации DDD в Python:
https://github.com/NEONKID/fastapi‑ddd‑example
https://github.com/cosmicpython/code