clean-architecture

Clean Architecture

Structure code into Domain, Application, and Infrastructure layers with strict dependency rules, ensuring business logic is isolated from frameworks and external services.

When to Use

• Starting a new module or microservice
• Refactoring code that mixes business logic with infrastructure
• Creating entities, value objects, use cases, or repositories
• Reviewing code for layer violations
• Setting up project structure for a new feature

Available Operations

1. Define domain entities with business rules and events
2. Create application use cases that orchestrate domain logic
3. Implement infrastructure adapters (repositories, HTTP clients)
4. Set up dependency injection wiring
5. Validate layer dependencies (no inward violations)

Multi-Step Workflow

Step 1: Set Up the Layer Structure

Create the directory structure following the dependency rule: outer layers depend on inner layers, never the reverse.

mkdir -p src/domain/entities src/domain/repositories src/domain/value_objects src/domain/events
mkdir -p src/application/use_cases src/application/dtos
mkdir -p src/infrastructure/persistence src/infrastructure/http
mkdir -p src/api/controllers

+------------------------------+
|         API / CLI            |  <- Controllers, Routers
+------------------------------+
|        APPLICATION           |  <- Use Cases, DTOs
+------------------------------+
|          DOMAIN              |  <- Entities, VOs, Events, Repos (interface)
+------------------------------+
|       INFRASTRUCTURE         |  <- DB, HTTP clients, Frameworks
+------------------------------+

Dependency Rule: arrows point INWARD (infra -> domain)
Domain NEVER imports from infrastructure

Step 2: Build the Domain Layer

Domain contains entities with business rules, value objects, domain events, and repository interfaces (ports).

# domain/entities/demand.py
class Demand:
    def __init__(self, id: DemandId, description: str):
        self._id = id
        self._description = description
        self._status = DemandStatus.CREATED
        self._events: list[DomainEvent] = []

    def decompose(self, planner: TaskPlanner) -> list[Task]:
        if self._status != DemandStatus.CREATED:
            raise DemandAlreadyDecomposedException(self._id)
        tasks = planner.plan(self._description)
        self._status = DemandStatus.PLANNED
        self._events.append(DemandDecomposed(self._id, [t.id for t in tasks]))
        return tasks

    @property
    def pending_events(self) -> list[DomainEvent]:
        return list(self._events)

# domain/repositories/demand_repository.py (PORT -- interface only)
from abc import ABC, abstractmethod

class DemandRepository(ABC):
    @abstractmethod
    def find_by_id(self, id: DemandId) -> Demand: ...
    @abstractmethod
    def save(self, demand: Demand) -> None: ...