Введение

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

Знакомо?

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

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

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

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;

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

# 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: ...

На практике интерфейс нужен не "на всякий случай". Абстракции обычно появляются там, где компонент неустойчивый и может иметь несколько реализаций, чаще всего это инфраструктурные зависимости: файловые хранилища, внешние сервисы и т.д. Например, для use case отдельный интерфейс почти всегда излишен - это устойчивая бизнес-логика. Интерфейс здесь только усложнит код и не принесёт пользы.

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

Доменные модели - это представление бизнес-сущности в коде. Их задача - описывать свойства объекта и его локальные правила.

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

# 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

Доменная модель отвечает за собственное состояние и локальные правила. Она работает только со своими данными и не зависит от инфраструктуры: БД, ORM, API, файловой системы, SDK и т.д. Координация сценариев и работа с внешними ресурсами обычно находятся на уровне use cases. Благодаря этому, изменения инфраструктуры - например, смена файлового хранилища или внешнего сервиса не влияют на доменный код и не тянут изменения через всё приложение.

Use Cases

Use case - это описание конкретного бизнес-сценария. Он описывает атомарный бизнес-процесс, связывая сущности, правила и взаимодействие с внешними ресурсами.

Для нашего примера это выглядит так:

1. Проверка состояния сущности - перед загрузкой инвойса проверяем, не отменён ли он. Это локальное правило самой сущности.

2. Взаимодействие с инфраструктурой через абстракции - Сохраняем файл через storage.save(key, content). Use case не зависит от конкретного способа хранения файла - это может быть локальная директория, S3, тестовый mock.

3. Обновление состояния сущности - после успешного сохранения файла вызываем invoice.mark_uploaded(). Инвойс отвечает только за изменение собственного состояния.

4. Сохраняем изменения в БД с помощью Unit of Work, бизнес-логика обычно не зависит от деталей ORM.

5. Use case возвращает объект с результатом - это готовый результат для внешнего слоя (HTTP API, CLI, event handler).

# 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()

          self.storage.save(str(invoice_id), content)

          invoice.mark_uploaded()

          self.uow.invoice_gate.save(invoice)

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

Подробнее про то как правильно их писать в следующем разделе - "Нюансы написания use cases на практике"

Адаптеры

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

Иными словами, use case просто говорит:

storage.save(key, content)

а адаптер решает как и где это сделать - локальная директория, S3, тестовый mock и т.д.

Вот адаптеры для нашего примера:

# 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 знает:

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

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

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

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

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

• и т.д.

P.S. Обычно адаптеры регистрируются в DI-контейнерах, чтобы use case получал их автоматически. Реализация DI выходит за рамки этой статьи, но если вам интересно, как это сделать на практике - оставьте комментарий, и я скину пример.

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

Слой представления - принимает внешний запрос, преобразует его в формат, понятный use case, и возвращает результат обратно в клиентский формат. Presentation layer занимается преобразованием форматов и коммуникацией с внешним миром, он не содержит бизнес-логики и не знает, где и как хранится файл.

В примере мы принимаем HTTP запрос с файлом, извлекаем нужные данные (invoice_id и содержимое файла), вызываем use case, и формируем HTTP-ответ

# 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 case из другого - это часто приводит к скрытой связности.

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

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

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

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

DTO и контракты данных

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

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

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

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

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

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

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

Формат ответа не должен быть общим на всю систему. Инфраструктура отдаёт данные как ей удобно, определенные use cases преобразуют их под свой сценарий, а слой представления - под свой. В итоге каждый слой делает свою адаптацию, не навязывая формат остальным.

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

Exceptions и контракты ошибок

С ошибками ситуация обычно проще - их редко приходится трансформировать между слоями так же активно, как DTO.

В системе обычно можно выделить несколько уровней ошибок: доменные, прикладные (use case) и инфраструктурные:

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

# domain/invoice/exceptions.py
class InvoiceCancelledError(Exception): ...


# 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

Дальше уже появляется второй тип - ошибки уровня use case (прикладные). Это ошибки не самой сущности, а бизнес-сценария:

# use_cases/exceptions/base.py
class ApplicationError(Exception): ...
# use_cases/exceptions/invoice.py
class UploadInvoiceFailedError(ApplicationError): ...


# usecases/invoice/upload.py
class UploadInvoiceUseCase:
    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()

            try:
                self.storage.save(
                    key=str(invoice.id),
                    content=content,
                )
            except StorageUnavailableError as e:
                raise UploadInvoiceFailedError(
                    "File storage is unavailable"
                ) from e

            invoice.mark_uploaded()

            self.uow.invoice_gate.save(invoice)

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

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

# contracts/files/storage.py
class StorageUnavailableError(Exception): ...

# adapters/files/storage/s3.py
class S3FileStorage(IFileStorage):
  # ...
    def save(self, key: str, content: bytes) -> None:
        try:
            self.s3.put_object(
                Bucket=self.bucket,
                Key=key,
                Body=content,
            )

        except botocore.exceptions.ClientError as e:
            raise StorageUnavailableError() from e

А Presentation layer это место, где система переводит ошибки в язык внешнего мира. Он в основном работает с доменными и прикладными ошибками. Вот пример для HTTP:

try:
    use_case.execute(...)
except InvoiceCancelledError:
    raise HTTPException(status.CONFLICT, "Invoice is cancelled")
except UploadInvoiceFailedError as e:
    raise HTTPException(status.BAD_REQUEST, str(e))

В итоге мы посмотрели, как ведут себя контракты данных и ошибок на границах слоёв, и где именно чаще всего появляется протекание между ними.

Заключение

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

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

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

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

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