clean-architecture

Clean Architecture, DDD & Hexagonal Architecture for Python

Overview

This skill provides comprehensive guidance for implementing Clean Architecture, Hexagonal Architecture (Ports & Adapters), and Domain-Driven Design patterns in Python applications. It focuses on creating maintainable, testable, and framework-independent business logic through proper separation of concerns.

Core Concepts

Layered Architecture (Clean Architecture) - Dependencies flow inward, inner layers know nothing about outer layers:

+-------------------------------------+
|  Infrastructure (Frameworks, DB)   |  <- Outer layer
+-------------------------------------+
|  Adapters (Controllers, Repos)     |
+-------------------------------------+
|  Use Cases (Application Logic)     |
+-------------------------------------+
|  Domain (Entities, Value Objects)  |  <- Inner layer
+-------------------------------------+

Layers:

• Domain: Entities, value objects, domain events, repository interfaces
• Use Cases: Application business rules, orchestrate domain objects
• Adapters: Interface implementations (controllers, repositories, gateways)
• Infrastructure: Framework configuration, database connections, external clients

Hexagonal Architecture (Ports & Adapters)

• Ports: Abstract interfaces defining what the application needs
• Adapters: Concrete implementations of ports
• Domain Core: Business logic with no external dependencies

Domain-Driven Design Tactical Patterns

• Entities: Objects with identity and lifecycle
• Value Objects: Immutable objects defined by attributes
• Aggregates: Consistency boundaries with aggregate roots
• Repositories: Persistence abstraction for aggregates
• Domain Events: Capture significant occurrences in the domain

When to Use

• Designing new Python backend systems with separation of concerns
• Refactoring tightly coupled code into layered architectures
• Implementing domain-driven design with bounded contexts
• Creating testable business logic independent of frameworks
• Building applications with FastAPI or Flask using clean patterns
• Setting up repository patterns with SQLAlchemy or async databases
• Implementing use case patterns with proper dependency injection

Instructions

1. Define the Project Structure

Create the layered directory structure following the dependency rule:

myapp/
+-- domain/                    # Inner layer - no external deps
|   +-- entities/             # Business entities
|   +-- value_objects/        # Immutable value objects
|   +-- events/               # Domain events
|   +-- repositories/         # Abstract repository interfaces (ports)
+-- use_cases/                # Application layer
+-- adapters/                 # Interface adapters
|   +-- repositories/         # Repository implementations
|   +-- controllers/          # API controllers
+-- infrastructure/           # Framework & external concerns
|   +-- database.py          # Database configuration
|   +-- container.py         # Dependency injection container
|   +-- config.py            # Application settings
+-- main.py                  # Application entry point

2. Implement the Domain Layer

Start from the innermost layer with no external dependencies:

1. Create Value Objects using frozen dataclasses with validation in __post_init__
2. Define Entities with identity, behavior, and factory methods (e.g., create())
3. Define Repository Interfaces (Ports) as abstract base classes with abstract methods
4. Keep all domain logic in entities - avoid anemic models

3. Implement the Use Cases Layer

Create application-specific business rules:

1. Define Request/Response dataclasses for input/output
2. Create Use Case classes that receive repository interfaces via constructor injection
3. Implement the execute() method that orchestrates domain objects
4. Handle validation and business errors, returning appropriate responses

4. Implement the Adapter Layer

Create concrete implementations of domain interfaces:

1. Implement Repository classes that extend domain interfaces
2. Use SQLAlchemy async sessions or other ORM tools
3. Map between domain entities and database models
4. Create Controllers (FastAPI routers) that invoke use cases

5. Implement the Infrastructure Layer

Configure frameworks and external dependencies:

1. Set up database connections and session management
2. Configure the dependency injection container
3. Wire all components together
4. Define application settings and configuration

6. Create the Application Entry Point

Build the FastAPI or Flask application:

1. Initialize the DI container and wire modules
2. Configure application lifespan (startup/shutdown)
3. Register routers and middleware
4. Export the application factory function

7. Write Tests

Test each layer in isolation:

1. Unit test use cases with mocked repositories
2. Unit test domain entities and value objects
3. Integration test adapters with test databases
4. End-to-end test the full application stack

Examples

Example 1: Domain Layer - Value Object & Entity

# domain/value_objects/email.py
from dataclasses import dataclass
import re

@dataclass(frozen=True)
class Email:
    value: str
    def __post_init__(self):
        if not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', self.value):
            raise ValueError(f"Invalid email: {self.value}")
    def __str__(self) -> str:
        return self.value


# domain/entities/user.py
from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID, uuid4
from domain.value_objects.email import Email

@dataclass
class User:
    email: Email
    name: str
    id: UUID = field(default_factory=uuid4)
    is_active: bool = True
    created_at: datetime = field(default_factory=datetime.utcnow)

    def deactivate(self) -> None:
        self.is_active = False

    def can_login(self) -> bool:
        return self.is_active

    @classmethod
    def create(cls, email: Email, name: str) -> "User":
        return cls(email=email, name=name)

Example 2: Repository Port (Interface)

# domain/repositories/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
from uuid import UUID
from domain.entities.user import User
from domain.value_objects.email import Email

class IUserRepository(ABC):
    @abstractmethod
    async def find_by_id(self, user_id: UUID) -> Optional[User]: ...
    @abstractmethod
    async def find_by_email(self, email: Email) -> Optional[User]: ...
    @abstractmethod
    async def save(self, user: User) -> User: ...
    @abstractmethod
    async def delete(self, user_id: UUID) -> bool: ...

Example 3: Use Case Layer

# use_cases/create_user.py
from dataclasses import dataclass
from typing import Optional
from uuid import UUID
from domain.entities.user import User
from domain.value_objects.email import Email
from domain.repositories.user_repository import IUserRepository

@dataclass
class CreateUserRequest:
    email: str
    name: str

@dataclass
class CreateUserResponse:
    user_id: Optional[UUID]
    success: bool
    error_message: Optional[str] = None

class CreateUserUseCase:
    def __init__(self, user_repository: IUserRepository):
        self._user_repository = user_repository

    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
        try:
            email = Email(request.email)
        except ValueError as e:
            return CreateUserResponse(None, False, str(e))
        if await self._user_repository.find_by_email(email):
            return CreateUserResponse(None, False, "Email already registered")
        user = User.create(email=email, name=request.name)
        saved = await self._user_repository.save(user)
        return CreateUserResponse(saved.id, True)

Example 4: Adapter Layer - Repository Implementation

# adapters/repositories/s