clean-architecture

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Clean Architecture Patterns

整洁架构模式

Build maintainable, testable backends with SOLID principles and hexagonal architecture.
基于SOLID原则与六边形架构构建可维护、可测试的后端系统。

SOLID Principles ( Python)

SOLID原则(Python实现)

S - Single Responsibility

S - 单一职责原则

python
undefined
python
undefined

BAD: One class doing everything

BAD: One class doing everything

class UserManager: def create_user(self, data): ... def send_welcome_email(self, user): ... def generate_report(self, users): ...
class UserManager: def create_user(self, data): ... def send_welcome_email(self, user): ... def generate_report(self, users): ...

GOOD: Separate responsibilities

GOOD: Separate responsibilities

class UserService: def create_user(self, data: UserCreate) -> User: ...
class EmailService: def send_welcome(self, user: User) -> None: ...
class ReportService: def generate_user_report(self, users: list[User]) -> Report: ...
undefined
class UserService: def create_user(self, data: UserCreate) -> User: ...
class EmailService: def send_welcome(self, user: User) -> None: ...
class ReportService: def generate_user_report(self, users: list[User]) -> Report: ...
undefined

O - Open/Closed (Protocol-based)

O - 开闭原则(基于Protocol实现)

python
from typing import Protocol

class PaymentProcessor(Protocol):
    async def process(self, amount: Decimal) -> PaymentResult: ...

class StripeProcessor:
    async def process(self, amount: Decimal) -> PaymentResult:
        # Stripe implementation
        ...

class PayPalProcessor:
    async def process(self, amount: Decimal) -> PaymentResult:
        # PayPal implementation - extends without modifying
        ...
python
from typing import Protocol

class PaymentProcessor(Protocol):
    async def process(self, amount: Decimal) -> PaymentResult: ...

class StripeProcessor:
    async def process(self, amount: Decimal) -> PaymentResult:
        # Stripe implementation
        ...

class PayPalProcessor:
    async def process(self, amount: Decimal) -> PaymentResult:
        # PayPal implementation - extends without modifying
        ...

L - Liskov Substitution

L - 里氏替换原则

python
undefined
python
undefined

Any implementation of Repository can substitute another

Any implementation of Repository can substitute another

class IUserRepository(Protocol): async def get_by_id(self, id: str) -> User | None: ... async def save(self, user: User) -> User: ...
class PostgresUserRepository: async def get_by_id(self, id: str) -> User | None: ... async def save(self, user: User) -> User: ...
class InMemoryUserRepository: # For testing - fully substitutable async def get_by_id(self, id: str) -> User | None: ... async def save(self, user: User) -> User: ...
undefined
class IUserRepository(Protocol): async def get_by_id(self, id: str) -> User | None: ... async def save(self, user: User) -> User: ...
class PostgresUserRepository: async def get_by_id(self, id: str) -> User | None: ... async def save(self, user: User) -> User: ...
class InMemoryUserRepository: # For testing - fully substitutable async def get_by_id(self, id: str) -> User | None: ... async def save(self, user: User) -> User: ...
undefined

I - Interface Segregation

I - 接口隔离原则

python
undefined
python
undefined

BAD: Fat interface

BAD: Fat interface

class IRepository(Protocol): async def get(self, id: str): ... async def save(self, entity): ... async def delete(self, id: str): ... async def search(self, query: str): ... async def bulk_insert(self, entities): ...
class IRepository(Protocol): async def get(self, id: str): ... async def save(self, entity): ... async def delete(self, id: str): ... async def search(self, query: str): ... async def bulk_insert(self, entities): ...

GOOD: Segregated interfaces

GOOD: Segregated interfaces

class IReader(Protocol): async def get(self, id: str) -> T | None: ...
class IWriter(Protocol): async def save(self, entity: T) -> T: ...
class ISearchable(Protocol): async def search(self, query: str) -> list[T]: ...
undefined
class IReader(Protocol): async def get(self, id: str) -> T | None: ...
class IWriter(Protocol): async def save(self, entity: T) -> T: ...
class ISearchable(Protocol): async def search(self, query: str) -> list[T]: ...
undefined

D - Dependency Inversion

D - 依赖倒置原则

python
from typing import Protocol
from fastapi import Depends

class IAnalysisRepository(Protocol):
    async def get_by_id(self, id: str) -> Analysis | None: ...

class AnalysisService:
    def __init__(self, repo: IAnalysisRepository):
        self._repo = repo  # Depends on abstraction, not concrete
python
from typing import Protocol
from fastapi import Depends

class IAnalysisRepository(Protocol):
    async def get_by_id(self, id: str) -> Analysis | None: ...

class AnalysisService:
    def __init__(self, repo: IAnalysisRepository):
        self._repo = repo  # Depends on abstraction, not concrete

FastAPI DI

FastAPI DI

def get_analysis_service( db: AsyncSession = Depends(get_db) ) -> AnalysisService: repo = PostgresAnalysisRepository(db) return AnalysisService(repo)
undefined
def get_analysis_service( db: AsyncSession = Depends(get_db) ) -> AnalysisService: repo = PostgresAnalysisRepository(db) return AnalysisService(repo)
undefined

Hexagonal Architecture (Ports & Adapters)

六边形架构(端口与适配器模式)

┌─────────────────────────────────────────────────────────────┐
│                      DRIVING ADAPTERS                        │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐    │
│  │ FastAPI  │  │   CLI    │  │  Celery  │  │  Tests   │    │
│  │ Routes   │  │ Commands │  │  Tasks   │  │  Mocks   │    │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘    │
│       │             │             │             │           │
│       ▼             ▼             ▼             ▼           │
│  ╔═══════════════════════════════════════════════════════╗  │
│  ║                    INPUT PORTS                        ║  │
│  ║  ┌─────────────────┐  ┌─────────────────────────────┐ ║  │
│  ║  │ AnalysisService │  │ UserService                 │ ║  │
│  ║  │ (Use Cases)     │  │ (Use Cases)                 │ ║  │
│  ║  └────────┬────────┘  └──────────────┬──────────────┘ ║  │
│  ╠═══════════╪══════════════════════════╪════════════════╣  │
│  ║           ▼          DOMAIN          ▼                ║  │
│  ║  ┌─────────────────────────────────────────────────┐  ║  │
│  ║  │  Entities  │  Value Objects  │  Domain Events   │  ║  │
│  ║  │  Analysis  │  AnalysisType   │  AnalysisCreated │  ║  │
│  ║  └─────────────────────────────────────────────────┘  ║  │
│  ╠═══════════════════════════════════════════════════════╣  │
│  ║                   OUTPUT PORTS                        ║  │
│  ║  ┌──────────────────┐  ┌────────────────────────────┐ ║  │
│  ║  │ IAnalysisRepo    │  │ INotificationService       │ ║  │
│  ║  │ (Protocol)       │  │ (Protocol)                 │ ║  │
│  ║  └────────┬─────────┘  └──────────────┬─────────────┘ ║  │
│  ╚═══════════╪══════════════════════════╪════════════════╝  │
│              ▼                          ▼                   │
│  ┌───────────────────┐  ┌────────────────────────────────┐ │
│  │ PostgresRepo      │  │ EmailNotificationService       │ │
│  │ (SQLAlchemy)      │  │ (SMTP/SendGrid)                │ │
│  └───────────────────┘  └────────────────────────────────┘ │
│                      DRIVEN ADAPTERS                        │
└─────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────┐
│                      驱动适配器 (DRIVING ADAPTERS)           │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐    │
│  │ FastAPI  │  │   CLI    │  │  Celery  │  │  Tests   │    │
│  │ 路由     │  │ 命令工具 │  │ 任务调度 │  │ 模拟测试 │    │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘    │
│       │             │             │             │           │
│       ▼             ▼             ▼             ▼           │
│  ╔═══════════════════════════════════════════════════════╗  │
│  ║                    输入端口 (INPUT PORTS)               ║  │
│  ║  ┌─────────────────┐  ┌─────────────────────────────┐ ║  │
│  ║  │ AnalysisService │  │ UserService                 │ ║  │
│  ║  │ (用例逻辑)     │  │ (用例逻辑)                 │ ║  │
│  ║  └────────┬────────┘  └──────────────┬──────────────┘ ║  │
│  ╠═══════════╪══════════════════════════╪════════════════╣  │
│  ║           ▼          领域层 (DOMAIN) ▼                ║  │
│  ║  ┌─────────────────────────────────────────────────┐  ║  │
│  ║  │ 实体       │  值对象       │  领域事件         │  ║  │
│  ║  │ Analysis   │ AnalysisType   │ AnalysisCreated │  ║  │
│  ║  └─────────────────────────────────────────────────┘  ║  │
│  ╠═══════════════════════════════════════════════════════╣  │
│  ║                   输出端口 (OUTPUT PORTS)               ║  │
│  ║  ┌──────────────────┐  ┌────────────────────────────┐ ║  │
│  ║  │ IAnalysisRepo    │  │ INotificationService       │ ║  │
│  ║  │ (协议定义)       │  │ (协议定义)                 │ ║  │
│  ║  └────────┬─────────┘  └──────────────┬─────────────┘ ║  │
│  ╚═══════════╪══════════════════════════╪════════════════╝  │
│              ▼                          ▼                   │
│  ┌───────────────────┐  ┌────────────────────────────────┐ │
│  │ PostgresRepo      │  │ EmailNotificationService       │ │
│  │ (SQLAlchemy实现) │  │ (SMTP/SendGrid实现)          │ │
│  └───────────────────┘  └────────────────────────────────┘ │
│                      被驱动适配器 (DRIVEN ADAPTERS)         │
└─────────────────────────────────────────────────────────────┘

DDD Tactical Patterns

DDD战术模式

Entity (Identity-based)

实体(基于标识)

python
from dataclasses import dataclass, field
from uuid import UUID, uuid4

@dataclass
class Analysis:
    id: UUID = field(default_factory=uuid4)
    source_url: str
    status: AnalysisStatus
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Analysis):
            return False
        return self.id == other.id  # Identity equality
python
from dataclasses import dataclass, field
from uuid import UUID, uuid4

@dataclass
class Analysis:
    id: UUID = field(default_factory=uuid4)
    source_url: str
    status: AnalysisStatus
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Analysis):
            return False
        return self.id == other.id  # 基于标识判断相等

Value Object (Structural equality)

值对象(基于结构相等性)

python
from dataclasses import dataclass

@dataclass(frozen=True)  # Immutable
class AnalysisType:
    category: str
    depth: int

    def __post_init__(self):
        if self.depth < 1 or self.depth > 3:
            raise ValueError("Depth must be 1-3")
python
from dataclasses import dataclass

@dataclass(frozen=True)  # 不可变对象
class AnalysisType:
    category: str
    depth: int

    def __post_init__(self):
        if self.depth < 1 or self.depth > 3:
            raise ValueError("深度必须在1-3之间")

Aggregate Root

聚合根

python
class AnalysisAggregate:
    def __init__(self, analysis: Analysis, artifacts: list[Artifact]):
        self._analysis = analysis
        self._artifacts = artifacts
        self._events: list[DomainEvent] = []

    def complete(self, summary: str) -> None:
        self._analysis.status = AnalysisStatus.COMPLETED
        self._analysis.summary = summary
        self._events.append(AnalysisCompleted(self._analysis.id))

    def collect_events(self) -> list[DomainEvent]:
        events = self._events.copy()
        self._events.clear()
        return events
python
class AnalysisAggregate:
    def __init__(self, analysis: Analysis, artifacts: list[Artifact]):
        self._analysis = analysis
        self._artifacts = artifacts
        self._events: list[DomainEvent] = []

    def complete(self, summary: str) -> None:
        self._analysis.status = AnalysisStatus.COMPLETED
        self._analysis.summary = summary
        self._events.append(AnalysisCompleted(self._analysis.id))

    def collect_events(self) -> list[DomainEvent]:
        events = self._events.copy()
        self._events.clear()
        return events

Directory Structure

目录结构

backend/app/
├── api/v1/              # Driving adapters (FastAPI routes)
├── domains/
│   └── analysis/
│       ├── entities.py      # Domain entities
│       ├── value_objects.py # Value objects
│       ├── services.py      # Domain services (use cases)
│       ├── repositories.py  # Output port protocols
│       └── events.py        # Domain events
├── infrastructure/
│   ├── repositories/    # Driven adapters (PostgreSQL)
│   ├── services/        # External service adapters
│   └── messaging/       # Event publishers
└── core/
    ├── dependencies.py  # FastAPI DI configuration
    └── protocols.py     # Shared protocols
backend/app/
├── api/v1/              # 驱动适配器(FastAPI路由)
├── domains/
│   └── analysis/
│       ├── entities.py      # 领域实体
│       ├── value_objects.py # 值对象
│       ├── services.py      # 领域服务(用例逻辑)
│       ├── repositories.py  # 输出端口协议
│       └── events.py        # 领域事件
├── infrastructure/
│   ├── repositories/    # 被驱动适配器(PostgreSQL实现)
│   ├── services/        # 外部服务适配器
│   └── messaging/       # 事件发布器
└── core/
    ├── dependencies.py  # FastAPI依赖注入配置
    └── protocols.py     # 共享协议定义

Anti-Patterns (FORBIDDEN)

反模式(禁止使用)

python
undefined
python
undefined

NEVER import infrastructure in domain

绝对不要在领域层导入基础设施代码

from app.infrastructure.database import engine # In domain layer
from app.infrastructure.database import engine # 错误:领域层中引用基础设施

NEVER leak ORM models to API

绝对不要将ORM模型暴露到API层

@router.get("/users/{id}") async def get_user(id: str, db: Session) -> UserModel: # Returns ORM model return db.query(UserModel).get(id)
@router.get("/users/{id}") async def get_user(id: str, db: Session) -> UserModel: # 错误:直接返回ORM模型 return db.query(UserModel).get(id)

NEVER have domain depend on framework

绝对不要让领域层依赖框架

from fastapi import HTTPException class UserService: def get(self, id: str): if not user: raise HTTPException(404) # Framework in domain!
undefined
from fastapi import HTTPException class UserService: def get(self, id: str): if not user: raise HTTPException(404) # 错误:领域层中使用框架异常
undefined

Key Decisions

关键决策

DecisionRecommendation
Protocol vs ABCUse Protocol (structural typing)
Dataclass vs PydanticDataclass for domain, Pydantic for API
Repository granularityOne per aggregate root
Transaction boundaryService layer, not repository
Event publishingCollect in aggregate, publish after commit
决策项推荐方案
Protocol vs ABC使用Protocol(结构类型)
数据类 vs Pydantic领域层用Dataclass,API层用Pydantic
仓储粒度每个聚合根对应一个仓储
事务边界服务层而非仓储层
事件发布聚合根中收集事件,提交后发布

Related Skills

相关技能

  • repository-patterns
    - Detailed repository implementations
  • api-design-framework
    - REST API patterns
  • database-schema-designer
    - Schema design
  • repository-patterns
    - 详细的仓储模式实现
  • api-design-framework
    - REST API设计模式
  • database-schema-designer
    - 数据库Schema设计

Capability Details

能力详情

solid-principles

solid-principles

Keywords: SOLID, single responsibility, open closed, liskov, interface segregation, dependency inversion Solves:
  • How do I apply SOLID principles in Python?
  • My classes are doing too much
关键词: SOLID、单一职责、开闭原则、里氏替换、接口隔离、依赖倒置 解决问题:
  • 如何在Python中应用SOLID原则?
  • 我的类职责过于臃肿

hexagonal-architecture

hexagonal-architecture

Keywords: hexagonal, ports and adapters, clean architecture, onion Solves:
  • How do I structure my FastAPI app?
  • How to separate infrastructure from domain?
关键词: 六边形架构、端口与适配器、整洁架构、洋葱架构 解决问题:
  • 如何构建FastAPI项目结构?
  • 如何将基础设施与领域层分离?

ddd-tactical

ddd-tactical

Keywords: entity, value object, aggregate, domain event, DDD Solves:
  • What's the difference between entity and value object?
  • How to design aggregates?
关键词: 实体、值对象、聚合根、领域事件、DDD 解决问题:
  • 实体和值对象的区别是什么?
  • 如何设计聚合根?