AgentSkillsCN

clean-architecture

为Python/FastAPI项目实施清洁架构提供指南。在新建模块、将现有代码重构为清洁架构、设计分层结构,或实现端口/适配器模式时使用。可通过“清洁架构”“六边形架构”“端口与适配器”“分层结构”“DIP”“依赖倒置”等关键词触发。

SKILL.md
--- frontmatter
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

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

Layer Responsibilities

LayerResponsibilityAllowed Dependencies
DomainBusiness rules, Entities, Value ObjectsNone (pure Python)
ApplicationUse Cases, OrchestrationDomain only
InfrastructureDB, External APIs, Framework adaptersDomain, Application
PresentationHTTP/CLI, Request/Response handlingApplication

Standard Directory Structure

code
app_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)

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

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

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

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

TypePatternExample
Port (data access){Entity}Repository or {Entity}GatewayUserRepository
Port (action){Action}erPasswordHasher, TokenGenerator
Adapter{Tech}{Port}PostgresUserRepository, BcryptHasher
Command Use Case{Verb}{Noun}InteractorCreateUserInteractor
Query Use Case{Verb}{Noun}QueryServiceListUsersQueryService
Value Object{Noun} (noun form)Email, UserId, Money

Reference Files

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__