---

name: clean-architecture description: Guide for implementing Clean Architecture in Python/FastAPI projects. Use when creating new modules, refactoring existing code to Clean Architecture, designing layer structures, or implementing Port/Adapter patterns. Triggers on "clean architecture", "hexagonal", "ports and adapters", "layer structure", "DIP", "dependency inversion".

Clean Architecture Implementation Guide

Quick Reference

Copy┌─────────────────────────────────────────────────────────────────┐
│                         Dependency Rule                          │
│          Dependencies ALWAYS point inward (to Domain)            │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   Presentation ──▶ Application ──▶ Domain ◀── Infrastructure    │
│   (Controllers)    (Use Cases)    (Entities)   (Adapters)       │
│                                                                  │
│   Infrastructure IMPLEMENTS Domain/Application Ports             │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Layer Responsibilities

Standard Directory Structure

Copyapp_name/
├── domain/
│   ├── entities/           # Aggregate roots, Entities
│   ├── value_objects/      # Immutable value types
│   ├── services/           # Domain services (stateless)
│   ├── ports/              # Domain-level abstractions
│   ├── enums/              # Domain enumerations
│   └── exceptions/         # Domain exceptions
│
├── application/
│   ├── commands/           # Write Use Cases (Interactors)
│   ├── queries/            # Read Use Cases (QueryServices)
│   ├── ports/              # Application-level abstractions
│   ├── dto/                # Data Transfer Objects
│   ├── services/           # Application services
│   └── exceptions/         # Application exceptions
│
├── infrastructure/
│   ├── adapters/           # Port implementations
│   ├── persistence/        # ORM, migrations
│   ├── integrations/       # External API clients
│   └── exceptions/         # Infrastructure exceptions
│
├── presentation/
│   └── http/
│       ├── controllers/    # FastAPI routers
│       ├── dependencies/   # FastAPI Depends
│       └── schemas/        # Pydantic request/response
│
└── setup/
    ├── config.py           # Settings
    └── dependencies.py     # DI wiring

Port & Adapter Pattern

Defining Ports (Interfaces)

Copy# application/ports/user_repository.py
from typing import Protocol

class UserRepository(Protocol):
    """Port: Abstract interface for user persistence"""

    async def get_by_id(self, user_id: UserId) -> User | None: ...
    async def save(self, user: User) -> None: ...
    async def exists_by_email(self, email: Email) -> bool: ...

Implementing Adapters

Copy# infrastructure/adapters/postgres_user_repository.py
class PostgresUserRepository:
    """Adapter: Concrete implementation using PostgreSQL"""

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

    async def get_by_id(self, user_id: UserId) -> User | None:
        stmt = select(UserModel).where(UserModel.id == user_id.value)
        result = await self._session.execute(stmt)
        row = result.scalar_one_or_none()
        return self._to_entity(row) if row else None

CQRS Pattern

Command (Write Operation)

Copy# application/commands/create_user.py
@dataclass(frozen=True)
class CreateUserCommand:
    email: str
    password: str

class CreateUserInteractor:
    def __init__(
        self,
        user_repo: UserRepository,
        hasher: PasswordHasher,
        tx: TransactionManager,
    ):
        self._repo = user_repo
        self._hasher = hasher
        self._tx = tx

    async def execute(self, cmd: CreateUserCommand) -> UserId:
        # Domain logic
        user = User.create(
            email=Email(cmd.email),
            password_hash=await self._hasher.hash(cmd.password),
        )
        await self._repo.save(user)
        await self._tx.commit()
        return user.id

Query (Read Operation)

Copy# application/queries/get_user.py
class GetUserQueryService:
    def __init__(self, reader: UserQueryGateway):
        self._reader = reader

    async def execute(self, user_id: UUID) -> UserDTO | None:
        return await self._reader.get_by_id(user_id)

Naming Conventions

Reference Files

• Layer details: See layer-structure.md
• Port/Adapter examples: See port-adapter.md
• CQRS patterns: See cqrs-patterns.md
• Evaluation checklist: See evaluation-checklist.md
• Anti-patterns to avoid: See anti-patterns.md

Eco² Project Conventions

This project follows conventions from docs/foundations/16-fastapi-clean-example-analysis.md:

• Use Protocol over ABC for interfaces (structural typing)
• Gateway naming for CQRS split (CommandGateway, QueryGateway)
• Separate Flusher and TransactionManager ports
• Value Objects with self-validation in __post_init__