clean-architecture
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseClean Architecture Patterns
整洁架构模式
Build maintainable, testable backends with SOLID principles and hexagonal architecture.
基于SOLID原则与六边形架构构建可维护、可测试的后端系统。
SOLID Principles ( Python)
SOLID原则(Python实现)
S - Single Responsibility
S - 单一职责原则
python
undefinedpython
undefinedBAD: 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: ...
undefinedclass 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: ...
undefinedO - 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
undefinedpython
undefinedAny 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: ...
undefinedclass 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: ...
undefinedI - Interface Segregation
I - 接口隔离原则
python
undefinedpython
undefinedBAD: 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]: ...
undefinedclass 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]: ...
undefinedD - 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 concretepython
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 concreteFastAPI DI
FastAPI DI
def get_analysis_service(
db: AsyncSession = Depends(get_db)
) -> AnalysisService:
repo = PostgresAnalysisRepository(db)
return AnalysisService(repo)
undefineddef get_analysis_service(
db: AsyncSession = Depends(get_db)
) -> AnalysisService:
repo = PostgresAnalysisRepository(db)
return AnalysisService(repo)
undefinedHexagonal 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 equalitypython
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 eventspython
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 eventsDirectory 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 protocolsbackend/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
undefinedpython
undefinedNEVER 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!
undefinedfrom fastapi import HTTPException
class UserService:
def get(self, id: str):
if not user:
raise HTTPException(404) # 错误:领域层中使用框架异常
undefinedKey Decisions
关键决策
| Decision | Recommendation |
|---|---|
| Protocol vs ABC | Use Protocol (structural typing) |
| Dataclass vs Pydantic | Dataclass for domain, Pydantic for API |
| Repository granularity | One per aggregate root |
| Transaction boundary | Service layer, not repository |
| Event publishing | Collect in aggregate, publish after commit |
| 决策项 | 推荐方案 |
|---|---|
| Protocol vs ABC | 使用Protocol(结构类型) |
| 数据类 vs Pydantic | 领域层用Dataclass,API层用Pydantic |
| 仓储粒度 | 每个聚合根对应一个仓储 |
| 事务边界 | 服务层而非仓储层 |
| 事件发布 | 聚合根中收集事件,提交后发布 |
Related Skills
相关技能
- - Detailed repository implementations
repository-patterns - - REST API patterns
api-design-framework - - Schema design
database-schema-designer
- - 详细的仓储模式实现
repository-patterns - - REST API设计模式
api-design-framework - - 数据库Schema设计
database-schema-designer
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
解决问题:
- 实体和值对象的区别是什么?
- 如何设计聚合根?