Введение

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

Знакомо?

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

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

Посмотрите на функцию загрузки инвойса:

def upload_invoice(session: Session, base_path: str, invoice_id: UUID, content: bytes) -> str:
    file_path = f"{base_path}/{invoice_id}.pdf"

    with open(file_path, "wb") as f:
        f.write(content)

    invoice = session.query(InvoiceORM).filter(InvoiceORM.id == invoice_id).one()
    invoice.file_path = file_path
    invoice.status = "uploaded"

    session.commit()

    return file_path

Что тут не так, помимо отсутствия примитивов синхронизации?

Функция одновременно:

• работает с файловой системой;

• напрямую зависит от ORM;

Пока проект маленький - это кажется очень удобным. Но со временем любая инфрастуктурная задача начинает тянуть изменения через всё приложение.

Допустим, через некоторое время проект начинает расти и вам нужно переехать на S3 хранилище. Приходится писать еще одну функцию или еще хуже - переписывать старую:

def upload_invoice_s3(
    session: Session,
    s3: S3Client,
    bucket: str,
    key: str,
    invoice_id: UUID,
    content: bytes
) -> str:

    s3.put_object(Bucket=bucket, Key=key, Body=content)

    invoice = session.query(InvoiceORM).filter(InvoiceORM.id == invoice_id).one()
    invoice.file_key = key
    invoice.storage_type = "s3"
    invoice.status = "uploaded"

    session.commit()

    return f"//{bucket}.s3.amazonaws.com/{key}"

Но проблема уже глубже. Локальное файловое хранилище, скорее всего, используется по всему проекту:

• где-то напрямую открываются файлы;

• где-то собираются file_path;

• где-то проверяется существование файлов;

• где-то логика начинает зависеть от структуры директорий.

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

А на следующий день приходит задача:

Для локальной разработки нужно использовать файловую систему!

Получается, что инфраструктурная деталь начинает определять структуру бизнес кода.

Какое решение?

Если инфраструктурный компонент может меняться независимо от бизнес логики, имеет смысл вынести его за контракт. Бизнес-логика должна работать не с S3 или локальной директорией напрямую, а с абстракцией:

storage.save(key, content)
storage.get(key)

Тогда use case вообще не знает:

• как он хранится;

• какой SDK используется;

• локальное хранилище или удаленное.

В production DI контейнере используется S3FileStorage, в dev DI контейнере - LocalFileStorage. Смена инфраструктуры превращается в изменение конфигурации, а не в рефакторинг всего приложения.

При этом важно понимать: Clean Architecture совсем не бесплатная абстракция, она:

• увеличивает количество кода;

• повышает порог входа;

• усложняет навигацию по проекту;

• требует командной дисциплины;

• замедляет разработку небольших приложений.

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

Пошаговое внедрение на практике

Давайте теперь посмотрим, как это выглядит на практике - на том же примере загрузки invoice. Попробуем постепенно разделить бизнес-логику и инфраструктуру так, чтобы смена файлового хранилища перестала тянуть рефакторинг через всё приложение.

Контракты

Контракты - это граница между бизнес логикой и внешним миром. Здесь обычно живут:

• интерфейсы

• инфраструктурные input/output DTO;

• инфраструктурные exceptions;

# contracts/files/storage.py
class IFileStorage(abc.ABC):
    @abc.abstractmethod
    def save(self, key: str, content: bytes) -> None: ...

    @abc.abstractmethod
    def get(self, key: str) -> bytes: ...

Важно понимать: интерфейс нужен не “на всякий случай”. Абстракция имеет смысл только тогда, когда компонент неустойчивый и существует несколько реализаций. Создавать интерфейс для каждого класса подряд - такой же анти паттерн, как и полное отсутствие абстракций.

Доменные модели

Доменная модель описывает бизнес-сущность и её локальные правила.

# domain/invoice/entities.py
@dataclass
class Invoice:
    id: UUID
    user_id: UUID
    amount: Decimal
    status: InvoiceStatus

    @property
    def is_paid(self) -> bool:
        return self.status == InvoiceStatus.PAID

    @property
    def is_cancelled(self) -> bool:
    	return self.status == InvoiceStatus.CANCELLED

    def mark_uploaded(self) -> None:
        if self.is_cancelled:
            raise InvoiceCancelledError()
        self.status = InvoiceStatus.UPLOADED

Важно понимать, что доменная модель заботится только о собственных бизнес правилах: допустимых переходах состояний, защите своего состояния. Она не обращается к базе данных и не управляет внешними ресурсами, такими как файлы или API(смотрим первый паттерн в GRASP). Всё взаимодействие с внешней инфраструктурой делается через use cases, которые используют эту модель.

Если пытаться вынести проверки состояния, инварианты или другие локальные бизнес-правила в отдельные функции, это приводит к нескольким проблемам:

• Снижение выразительности кода: вызов invoice.is_paid() читается почти как естественный язык, а is_paid(invoice) уже не так очевиден

• Раздувание кода и ухудшение читаемости: c ростом количества проверок и бизнес-правил код становится громоздким, трудным для понимания и поддержки.

• Повышение внутренней связанности системы

Use Cases

Use case - это описание конкретного бизнес-сценария.

# usecases/invoice/upload.py
class UploadInvoiceUseCase:
    def __init__(self, storage: IFileStorage, uow: IUoW):
        self.storage = storage
        self.uow = uow

    def execute(
        self,
        invoice_id: UUID,
        content: bytes,
    ) -> UploadInvoiceOutput:
        with self.uow:
          invoice = self.uow.invoice_gate.get_by_id(invoice_id)

          if invoice.is_cancelled:
              raise InvoiceCancelledError()

          key = f"invoices/{invoice_id}.pdf"

          self.storage.save(key, content)

          invoice.mark_uploaded()

          self.uow.invoice_gate.save(invocie)
          self.uow.commit()

          return UploadInvoiceOutput(
              invoice_id=invoice.id,
              uploaded=True,
          )

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

Адаптеры

Адаптеры - это инфраструктурный слой. Именно здесь находятся конкретные реализации контрактов: PostgresQL, Kafka, SMTP, HTTP, StripeGateway, etc.

# adapters/files/storage/s3.py
class S3FileStorage(IFileStorage):
    def __init__(self, s3: S3Client, bucket: str):
        self.s3 = s3
        self.bucket = bucket

    def save(self, key: str, content: bytes) -> None:
        # конкретная реализация

    def get(self, key: str) -> bytes:
        # конкретная реализация


# adapters/files/storage/local.py
class LocalFileStorage(IFileStorage):
    def __init__(self, base_path: str):
        self.base_path = base_path

    def save(self, key: str, content: bytes) -> None:
        # конкретная реализация

    def get(self, key: str) -> bytes:
        # конкретная реализация

Именно adapter знает:

• как устроены библиотеки;

• какие примитивы синхронизации использовать;

• лимиты и особенности конкретного провайдера

• как управлять транзакцией;

• стратегии повторов, таймауты и как вести себя при деградации;

• и т.д.

Слой представления

Presentation layer - самая внешняя часть системы. На практике это обычно transport handler: http api, cli, message broker consumer, etc.

Его задача принять запрос, преобразовать данные, вызвать юскейс и вернуть ответ:

# handlers/api/v1/invoice/routes.py
@router.post("/invoices/{invoice_id}/upload", response_model=UploadInvoiceResponse)
@inject
def upload_invoice(
    invoice_id: UUID,
    file: UploadFile,
    use_case: FromInjector[UploadInvoiceUseCase],
) -> UploadInvoiceResponse:
    result = use_case.execute(
        invoice_id=invoice_id,
        content=file.file.read(),
    )
    return UploadInvoiceResponse(
        invoice_id=result.invoice_id,
        uploaded=result.uploaded,
    )

Нюансы написания Use Cases на практике

На практике почти всегда хочется “упростить жизнь” и собрать весь сценарий в один большой execute(). Например: создать инвойс, загрузить файл, отправить евент, обновить статистику.

Сначала это выглядит удобно. Есть один вход, один метод, один “бизнес-процесс”.

Проблемы начинаются позже. Например:

• появляется новый Actor, которому уведомления уже не нужны;

• появляется новый Actor, которому нужен batch processing;

• появляется новый Actor, которому нужна какая-то новая фича;

• или вообще появляется новый транспортный слой, у которого есть зависимость от внешних callbacks;

И со временем монолитный use case начинает зависеть от контекста вызова.

Поэтому на практике я стараюсь придерживаться простого правила: один use case - один атомарный бизнес-процесс, т.е мы объединяем в use case шаги, которые не имеют смысла по отдельности с точки зрения бизнес контекста вне этой операции. Но это правило не является абсолютным, есть исключения.

Это даёт несколько важных преимуществ:

• use cases остаются переиспользуемыми;

• они проще тестируются;

• уменьшается связанность системы.

Также в большинстве случаев стоит избегать вызова одного use case из другого - это часто приводит к скрытой связности.

Если знаете другие подходы к написанию use cases, которые хорошо работают на практике, буду очень благодарен за ваш опыт!

Контракты и границы слоёв

Когда говорят про Clean Architecture, обычно фокусируются на направлении зависимостей: domain не зависит от infrastructure, use cases не знают про framework.

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

Обычно это происходит незаметно. Сначала инфрастуктурные DTO начинают использоваться как результат выполнения use case, затем транспортный слой просто пробрасывает его дальше и всё работает, первый актор доволен - контракт идеально подходит под его сценарий.

Проблема появляется позже.

Появляется второй актор, которому этот же ответ уже не подходит:

• часть полей лишняя;

• формат не удобен;

• нужны дополнительные данные;

• структура ответа должна выглядеть иначе.

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

Формат ответа должен оставаться локальным для конкретного сценария, а адаптация под нужды конкретного актора - происходить там, где эта потребность возникает.

Заключение

Clean Architecture - это точно не обязательный стандарт для любого проекта. Если у вас небольшой CRUD сервис без сложных интеграций, подобная архитектура вполне может оказаться избыточной.

Проблемы обычно начинаются позже.

Когда система растёт, появляются новые интеграции, внешние сервисы, отдельные команды. Именно тогда начинают проявляться последствия высокой связанности: инфраструктурные детали проникают в бизнес-код, изменения становятся всё менее локальными, а даже небольшие доработки начинают тянуть за собой каскадный рефакторинг системы. В этот момент архитектура перестаёт быть теорией и становится вопросом стоимости изменений. По сути, Clean Architecture - это попытка сделать такие изменения более управляемыми.

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

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

Возможно, в ближайшие годы это заметно изменит и отношение к чистой архитектуре? А какие у вас мысли по этому поводу? Делитесь, буду рад почитать!