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:

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

3. Implement the Use Cases Layer

Create application-specific business rules:

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

4. Implement the Adapter Layer

Create concrete implementations of domain interfaces:

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

5. Implement the Infrastructure Layer

Configure frameworks and external dependencies:

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

6. Create the Application Entry Point

Build the FastAPI or Flask application:

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

7. Write Tests

Test each layer in isolation:

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

Examples

Example 1: Domain Layer - Value Object & Entity

python

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

python

# 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

python

# 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

python

# adapters/repositories/sqlalchemy_user_repository.py
from typing import Optional
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from domain.entities.user import User
from domain.value_objects.email import Email
from domain.repositories.user_repository import IUserRepository

class SQLAlchemyUserRepository(IUserRepository):
    def __init__(self, session: AsyncSession):
        self._session = session

    async def find_by_id(self, user_id: UUID) -> Optional[User]:
        result = await self._session.execute(
            select(UserModel).where(UserModel.id == user_id)
        )
        row = result.scalar_one_or_none()
        return self._to_entity(row) if row else None

    async def find_by_email(self, email: Email) -> Optional[User]:
        result = await self._session.execute(
            select(UserModel).where(UserModel.email == str(email))
        )
        row = result.scalar_one_or_none()
        return self._to_entity(row) if row else None

    async def save(self, user: User) -> User:
        model = UserModel(
            id=user.id, email=str(user.email), name=user.name,
            is_active=user.is_active, created_at=user.created_at
        )
        self._session.add(model)
        await self._session.commit()
        return user

    def _to_entity(self, model) -> User:
        return User(
            id=model.id, email=Email(model.email), name=model.name,
            is_active=model.is_active, created_at=model.created_at
        )

Example 5: Dependency Injection Container

python

# infrastructure/container.py
from dependency_injector import containers, providers
from adapters.repositories.sqlalchemy_user_repository import SQLAlchemyUserRepository
from use_cases.create_user import CreateUserUseCase
from infrastructure.database import get_session

class Container(containers.DeclarativeContainer):
    db_session = providers.Factory(get_session)
    user_repository = providers.Factory(SQLAlchemyUserRepository, session=db_session)
    create_user_use_case = providers.Factory(
        CreateUserUseCase, user_repository=user_repository
    )

Example 6: FastAPI Controller

python

# adapters/controllers/user_controller.py
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, EmailStr
from use_cases.create_user import CreateUserUseCase, CreateUserRequest
from infrastructure.container import Container
from dependency_injector.wiring import inject, Provide

router = APIRouter(prefix="/users", tags=["users"])

class CreateUserInput(BaseModel):
    email: EmailStr
    name: str

@router.post("/", status_code=status.HTTP_201_CREATED)
@inject
async def create_user(
    data: CreateUserInput,
    use_case: CreateUserUseCase = Depends(Provide[Container.create_user_use_case])
):
    request = CreateUserRequest(email=data.email, name=data.name)
    response = await use_case.execute(request)
    if not response.success:
        raise HTTPException(status_code=400, detail=response.error_message)
    return {"id": str(response.user_id)}

Example 7: Application Entry Point

python

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from adapters.controllers import user_controller
from infrastructure.container import Container
from infrastructure.database import init_db

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield

def create_app() -> FastAPI:
    container = Container()
    container.wire(modules=[user_controller])
    app = FastAPI(title="Clean Architecture API", lifespan=lifespan)
    app.container = container
    app.include_router(user_controller.router)
    return app

app = create_app()

Example 8: Unit Testing Use Cases

python

# tests/unit/test_create_user_use_case.py
import pytest
from unittest.mock import AsyncMock
from use_cases.create_user import CreateUserUseCase, CreateUserRequest
from domain.entities.user import User
from domain.value_objects.email import Email

@pytest.fixture
def mock_repository():
    return AsyncMock()

@pytest.fixture
def use_case(mock_repository):
    return CreateUserUseCase(user_repository=mock_repository)

@pytest.mark.asyncio
async def test_create_user_success(use_case, mock_repository):
    mock_repository.find_by_email.return_value = None
    mock_repository.save.return_value = User(
        email=Email("test@example.com"), name="Test User"
    )
    request = CreateUserRequest(email="test@example.com", name="Test User")
    response = await use_case.execute(request)
    assert response.success is True
    assert response.user_id is not None

@pytest.mark.asyncio
async def test_create_user_duplicate_email(use_case, mock_repository):
    mock_repository.find_by_email.return_value = AsyncMock()
    request = CreateUserRequest(email="test@example.com", name="Test User")
    response = await use_case.execute(request)
    assert response.success is False
    assert "already registered" in response.error_message

Best Practices

Dependency Rule: Dependencies must always point inward toward the domain - never outward
Immutable Value Objects: Always use frozen dataclasses for value objects with validation in __post_init__
Rich Domain Models: Put business logic in entities, not in services or use cases
Use Cases as Orchestrators: Use cases coordinate workflows but domain objects make decisions
Async by Default: Use async/await for all I/O operations to support modern async frameworks
Pydantic at Boundary: Use Pydantic models only at the API boundary, never in domain layer
Repository per Aggregate: Create one repository per aggregate root, not per entity
Factory Methods: Use @classmethod factory methods like create() for entity construction with invariants
Dependency Injection: Inject dependencies through constructors for testability
Structured Responses: Return structured response objects from use cases, not raw entities

Constraints and Warnings

Architecture Constraints

Dependency Rule: Dependencies must always point inward toward the domain - never outward
Framework Independence: Domain layer must have no framework dependencies (no FastAPI, SQLAlchemy, Pydantic imports)
Interface Segregation: Keep repository interfaces focused and small - avoid god interfaces
Repository per Aggregate: Create one repository per aggregate root, not per entity

Implementation Constraints

Immutable Value Objects: Always use frozen dataclasses for value objects
Rich Domain Models: Put business logic in entities, not in services or use cases
Use Cases as Orchestrators: Use cases coordinate workflows but domain objects make decisions
Async by Default: Use async/await for all I/O operations to support modern async frameworks
Pydantic at Boundary: Use Pydantic models only at the API boundary, not in domain layer

Common Pitfalls to Avoid

Anemic Domain Models: Entities with only getters/setters and no behavior violate DDD principles
Leaky Abstractions: ORM models leaking into domain layer creates tight coupling
Fat Controllers: Business logic in controllers instead of use cases defeats the architecture
Missing Abstractions: Direct database calls in use cases break the dependency rule
Circular Dependencies: Be careful with imports between layers - use dependency injection to avoid
Over-Engineering: Not every CRUD app needs full DDD - evaluate complexity before applying

References

references/python-clean-architecture.md - Python-specific patterns including Result type, Specification pattern, Event Bus, and manual DI
references/fastapi-implementation.md - Complete FastAPI example with middleware, Docker setup, and integration tests

Search AI Tools

clean-architecture

Install this agent skill to your Project

SKILL.md

Clean Architecture, DDD & Hexagonal Architecture for Python

Overview

Core Concepts

When to Use

Instructions

1. Define the Project Structure

2. Implement the Domain Layer

3. Implement the Use Cases Layer

4. Implement the Adapter Layer

5. Implement the Infrastructure Layer

6. Create the Application Entry Point

7. Write Tests

Examples

Example 1: Domain Layer - Value Object & Entity

Example 2: Repository Port (Interface)

Example 3: Use Case Layer

Example 4: Adapter Layer - Repository Implementation

Example 5: Dependency Injection Container

Example 6: FastAPI Controller

Example 7: Application Entry Point

Example 8: Unit Testing Use Cases

Best Practices

Constraints and Warnings

Architecture Constraints

Implementation Constraints

Common Pitfalls to Avoid

References