用于整洁架构的 Python 设计模式
使用SOLID设计模式构建可维护的Python应用程序
Clean Architecture 通过强调关注点分离和依赖管理，彻底改变了开发人员构建可扩展、可维护应用程序的方式。
在 Python 中，这些原则与语言的动态特性相结合，创建了灵活、可测试的系统，这些系统可以随着业务需求的变化而演变，而不会变成技术债务。
理解 Python 中的 Clean Architecture
Clean Architecture 由 Robert C. Martin（Uncle Bob）提出，将软件组织成同心层，其中依赖关系指向核心业务逻辑。这种架构模式确保了应用程序的关键业务规则独立于框架、数据库和外部服务。
核心理念
基本原理简单而强大：业务逻辑不应依赖基础设施。您的领域实体、用例和业务规则应能独立于使用 PostgreSQL 还是 MongoDB、FastAPI 还是 Flask、AWS 还是 Azure。
在 Python 中，这种理念与语言的“鸭子类型”和面向协议的编程完美契合，允许在不使用静态类型语言所需的仪式感的情况下实现清晰的分离。
Clean Architecture 的四层
实体层（领域）：具有企业级业务规则的纯业务对象。这些是 POJO（Plain Old Python Objects），没有外部依赖。
用例层（应用）：特定于应用的业务规则，协调实体和外部服务之间的数据流。
接口适配器层：在使用案例和实体最方便的格式与外部机构所需的格式之间转换数据。
框架与驱动层：所有外部细节，如数据库、Web 框架和外部 API。
Python 中的 SOLID 原则
SOLID 原则构成了 Clean Architecture 的基础。让我们探讨每个原则在 Python 中是如何体现的。如需全面了解 Python 中的设计模式，请参阅 Python 设计模式指南。
单一职责原则（SRP）
每个类应只有一个更改的原因：
# 不好：多个职责
class UserManager:
def create_user(self, user_data):
# 创建用户
pass
def send_welcome_email(self, user):
# 发送邮件
pass
def log_creation(self, user):
# 记录日志
pass
# 好：职责分离
class UserService:
def __init__(self, repository, email_service, logger):
self.repository = repository
self.email_service = email_service
self.logger = logger
def create_user(self, user_data):
user = User(**user_data)
self.repository.save(user)
self.email_service.send_welcome(user)
self.logger.info(f"用户创建: {user.id}")
return user
开放/封闭原则（OCP）
软件实体应对扩展开放，但对修改关闭：
from abc import ABC, abstractmethod
from typing import Protocol
# 使用 Protocol（Python 3.8+）
class PaymentProcessor(Protocol):
def process_payment(self, amount: float) -> bool:
...
class CreditCardProcessor:
def process_payment(self, amount: float) -> bool:
# 信用卡逻辑
return True
class PayPalProcessor:
def process_payment(self, amount: float) -> bool:
# PayPal 逻辑
return True
# 无需修改现有代码即可轻松扩展
class CryptoProcessor:
def process_payment(self, amount: float) -> bool:
# 加密货币逻辑
return True
里氏替换原则（LSP）
对象应能被其子类型替换而不破坏程序：
from abc import ABC, abstractmethod
class DataStore(ABC):
@abstractmethod
def save(self, key: str, value: str) -> None:
pass
@abstractmethod
def get(self, key: str) -> str:
pass
class PostgreSQLStore(DataStore):
def save(self, key: str, value: str) -> None:
# PostgreSQL 实现
pass
def get(self, key: str) -> str:
# PostgreSQL 实现
return ""
class RedisStore(DataStore):
def save(self, key: str, value: str) -> None:
# Redis 实现
pass
def get(self, key: str) -> str:
# Redis 实现
return ""
# 两者可以互换使用
def process_data(store: DataStore, key: str, value: str):
store.save(key, value)
return store.get(key)
接口隔离原则（ISP）
客户端不应被迫依赖其不使用的接口：
# 不好：臃肿的接口
class Worker(ABC):
@abstractmethod
def work(self): pass
@abstractmethod
def eat(self): pass
@abstractmethod
def sleep(self): pass
# 好：隔离的接口
class Workable(Protocol):
def work(self) -> None: ...
class Eatable(Protocol):
def eat(self) -> None: ...
class Human:
def work(self) -> None:
print("工作")
def eat(self) -> None:
print("吃饭")
class Robot:
def work(self) -> None:
print("工作")
# 不需要 eat 方法
依赖倒置原则（DIP）
高层模块不应依赖于低层模块。两者都应依赖于抽象：
from typing import Protocol
# 抽象
class EmailSender(Protocol):
def send(self, to: str, subject: str, body: str) -> None:
...
# 低层模块
class SMTPEmailSender:
def send(self, to: str, subject: str, body: str) -> None:
# SMTP 实现
pass
# 高层模块依赖于抽象
class UserRegistrationService:
def __init__(self, email_sender: EmailSender):
self.email_sender = email_sender
def register(self, email: str, name: str):
# 注册逻辑
self.email_sender.send(
to=email,
subject="欢迎！",
body=f"你好 {name}"
)
仓储模式：抽象数据访问
仓储模式为访问领域对象提供类似集合的接口，隐藏数据存储的细节。
基本仓储实现
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4
@dataclass
class User:
id: UUID
email: str
name: str
is_active: bool = True
class UserRepository(ABC):
@abstractmethod
def save(self, user: User) -> User:
pass
@abstractmethod
def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
@abstractmethod
def get_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
def list_all(self) -> List[User]:
pass
@abstractmethod
def delete(self, user_id: UUID) -> bool:
pass
SQLAlchemy 实现
from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
Base = declarative_base()
class UserModel(Base):
__tablename__ = 'users'
id = Column(PGUUID(as_uuid=True), primary_key=True)
email = Column(String, unique=True, nullable=False)
name = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
class SQLAlchemyUserRepository(UserRepository):
def __init__(self, session: Session):
self.session = session
def save(self, user: User) -> User:
user_model = UserModel(
id=user.id,
email=user.email,
name=user.name,
is_active=user.is_active
)
self.session.add(user_model)
self.session.commit()
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.id == user_id
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def get_by_email(self, email: str) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.email == email
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def list_all(self) -> List[User]:
users = self.session.query(UserModel).all()
return [
User(
id=u.id,
email=u.email,
name=u.name,
is_active=u.is_active
)
for u in users
]
def delete(self, user_id: UUID) -> bool:
result = self.session.query(UserModel).filter(
UserModel.id == user_id
).delete()
self.session.commit()
return result > 0
用于测试的内存仓储
class InMemoryUserRepository(UserRepository):
def __init__(self):
self.users: dict[UUID, User] = {}
def save(self, user: User) -> User:
self.users[user.id] = user
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
return self.users.get(user_id)
def get_by_email(self, email: str) -> Optional[User]:
for user in self.users.values():
if user.email == email:
return user
return None
def list_all(self) -> List[User]:
return list(self.users.values())
def delete(self, user_id: UUID) -> bool:
if user_id in self.users:
del self.users[user_id]
return True
return False
服务层：协调业务逻辑
服务层实现用例，并协调仓储、外部服务和领域逻辑之间的流程。
from typing import Optional
from uuid import uuid4
class UserAlreadyExistsError(Exception):
pass
class UserNotFoundError(Exception):
pass
class UserService:
def __init__(
self,
user_repository: UserRepository,
email_service: EmailSender,
event_publisher: 'EventPublisher'
):
self.user_repository = user_repository
self.email_service = email_service
self.event_publisher = event_publisher
def register_user(self, email: str, name: str) -> User:
# 检查用户是否存在
existing_user = self.user_repository.get_by_email(email)
if existing_user:
raise UserAlreadyExistsError(f"邮箱 {email} 的用户已存在")
# 创建新用户
user = User(
id=uuid4(),
email=email,
name=name,
is_active=True
)
# 保存到仓储
user = self.user_repository.save(user)
# 发送欢迎邮件
self.email_service.send(
to=user.email,
subject="欢迎！",
body=f"你好 {user.name}, 欢迎来到我们的平台！"
)
# 发布事件
self.event_publisher.publish('user.registered', {
'user_id': str(user.id),
'email': user.email
})
return user
def deactivate_user(self, user_id: UUID) -> User:
user = self.user_repository.get_by_id(user_id)
if not user:
raise UserNotFoundError(f"未找到用户 {user_id}")
user.is_active = False
user = self.user_repository.save(user)
self.event_publisher.publish('user.deactivated', {
'user_id': str(user.id)
})
return user
Python 中的依赖注入
Python 的动态特性使得依赖注入变得简单，无需使用重型框架。
构造函数注入
class OrderService:
def __init__(
self,
order_repository: 'OrderRepository',
payment_processor: PaymentProcessor,
notification_service: 'NotificationService'
):
self.order_repository = order_repository
self.payment_processor = payment_processor
self.notification_service = notification_service
def place_order(self, order_data: dict):
# 使用注入的依赖
pass
简单的依赖容器
from typing import Dict, Type, Callable, Any
class Container:
def __init__(self):
self._services: Dict[Type, Callable] = {}
self._singletons: Dict[Type, Any] = {}
def register(self, interface: Type, factory: Callable):
self._services[interface] = factory
def register_singleton(self, interface: Type, instance: Any):
self._singletons[interface] = instance
def resolve(self, interface: Type):
if interface in self._singletons:
return self._singletons[interface]
factory = self._services.get(interface)
if factory:
return factory(self)
raise ValueError(f"未找到 {interface} 的注册")
# 使用
def create_container() -> Container:
container = Container()
# 注册服务
container.register_singleton(
Session,
sessionmaker(bind=create_engine('postgresql://...'))()
)
container.register(
UserRepository,
lambda c: SQLAlchemyUserRepository(c.resolve(Session))
)
container.register(
EmailSender,
lambda c: SMTPEmailSender()
)
container.register(
UserService,
lambda c: UserService(
c.resolve(UserRepository),
c.resolve(EmailSender),
c.resolve(EventPublisher)
)
)
return container
六边形架构（端口和适配器）
六边形架构将业务逻辑置于中心，适配器处理外部通信。
定义端口（接口）
# 输入端口（主）
class CreateUserUseCase(Protocol):
def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
...
# 输出端口（次）
class UserPersistencePort(Protocol):
def save(self, user: User) -> User:
...
def find_by_email(self, email: str) -> Optional[User]:
...
实现适配器
from pydantic import BaseModel, EmailStr
# 输入适配器（REST API）
from fastapi import FastAPI, Depends, HTTPException
class CreateUserRequest(BaseModel):
email: EmailStr
name: str
class CreateUserResponse(BaseModel):
id: str
email: str
name: str
app = FastAPI()
@app.post("/users", response_model=CreateUserResponse)
def create_user(
request: CreateUserRequest,
user_service: UserService = Depends(get_user_service)
):
try:
user = user_service.register_user(
email=request.email,
name=request.name
)
return CreateUserResponse(
id=str(user.id),
email=user.email,
name=user.name
)
except UserAlreadyExistsError as e:
raise HTTPException(status_code=400, detail=str(e))
# 输出适配器（数据库）
# 已经实现为 SQLAlchemyUserRepository
领域驱动设计模式
值对象
由其属性定义的不可变对象：
from dataclasses import dataclass
from typing import Pattern
import re
@dataclass(frozen=True)
class Email:
value: str
EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
def __post_init__(self):
if not self.EMAIL_PATTERN.match(self.value):
raise ValueError(f"无效的邮箱: {self.value}")
def __str__(self):
return self.value
@dataclass(frozen=True)
class Money:
amount: float
currency: str
def __post_init__(self):
if self.amount < 0:
raise ValueError("金额不能为负数")
if self.currency not in ['USD', 'EUR', 'GBP']:
raise ValueError(f"不支持的货币: {self.currency}")
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("不能相加不同货币")
return Money(self.amount + other.amount, self.currency)
聚合
被当作单一单元处理的领域对象集群：
from dataclasses import dataclass, field
from typing import List
from datetime import datetime
@dataclass
class OrderItem:
product_id: UUID
quantity: int
price: Money
def total(self) -> Money:
return Money(
self.price.amount * self.quantity,
self.price.currency
)
@dataclass
class Order:
id: UUID
customer_id: UUID
items: List[OrderItem] = field(default_factory=list)
status: str = "pending"
created_at: datetime = field(default_factory=datetime.now)
def add_item(self, product_id: UUID, quantity: int, price: Money):
item = OrderItem(product_id, quantity, price)
self.items.append(item)
def remove_item(self, product_id: UUID):
self.items = [
item for item in self.items
if item.product_id != product_id
]
def total(self) -> Money:
if not self.items:
return Money(0, "USD")
return sum(
(item.total() for item in self.items),
Money(0, self.items[0].price.currency)
)
def confirm(self):
if not self.items:
raise ValueError("无法确认空订单")
if self.status != "pending":
raise ValueError("订单已处理")
self.status = "confirmed"
领域事件
领域事件使组件之间松耦合，并支持事件驱动架构。对于生产级事件驱动系统，考虑使用如 AWS Kinesis 等服务实现事件流——参阅 使用 AWS Kinesis 构建事件驱动的微服务 获取详细指南。
from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable
@dataclass
class DomainEvent:
occurred_at: datetime = field(default_factory=datetime.now)
@dataclass
class OrderConfirmed(DomainEvent):
order_id: UUID
customer_id: UUID
total: Money
class EventPublisher:
def __init__(self):
self._handlers: Dict[Type, List[Callable]] = {}
def subscribe(self, event_type: Type, handler: Callable):
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event: DomainEvent):
event_type = type(event)
handlers = self._handlers.get(event_type, [])
for handler in handlers:
handler(event)
现代 Python 特性用于整洁架构
Python 的现代特性使实现整洁架构更加优雅且类型安全。如果你需要 Python 语法和特性的快速参考，请查看 Python 快速参考。
类型提示和协议
from typing import Protocol, runtime_checkable
@runtime_checkable
class Serializable(Protocol):
def to_dict(self) -> dict:
...
@classmethod
def from_dict(cls, data: dict) -> 'Serializable':
...
def serialize(obj: Serializable) -> dict:
return obj.to_dict()
使用 Pydantic 进行验证
from pydantic import BaseModel, Field, validator
from typing import Optional
class CreateUserDTO(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
@validator('name')
def name_must_not_contain_numbers(cls, v):
if any(char.isdigit() for char in v):
raise ValueError('Name cannot contain numbers')
return v
class Config:
frozen = True # 使其不可变
使用 Async/Await 进行 I/O 操作
Python 的 async/await 语法在整洁架构中对 I/O 密集型操作特别强大，允许与数据库和外部服务进行非阻塞交互。在将 Python 应用程序部署到无服务器平台时，了解性能特征变得至关重要——请参阅 AWS Lambda 性能：JavaScript vs Python vs Golang，以了解优化 Python 无服务器函数的见解。
from typing import List
import asyncio
class AsyncUserRepository(ABC):
@abstractmethod
async def save(self, user: User) -> User:
pass
@abstractmethod
async def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
class AsyncUserService:
def __init__(self, repository: AsyncUserRepository):
self.repository = repository
async def register_user(self, email: str, name: str) -> User:
user = User(id=uuid4(), email=email, name=name)
return await self.repository.save(user)
async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
tasks = [self.repository.get_by_id(uid) for uid in user_ids]
results = await asyncio.gather(*tasks)
return [u for u in results if u is not None]
项目结构最佳实践
良好的项目组织对于维护整洁架构至关重要。在设置项目结构之前，请确保使用 Python 虚拟环境进行依赖隔离。venv 快速参考 涵盖了管理虚拟环境所需了解的所有内容。对于现代 Python 项目，建议使用 uv - 新的 Python 包、项目和环境管理器，它提供了更快的包管理和项目设置。
my_application/
├── domain/ # 企业业务规则
│ ├── __init__.py
│ ├── entities/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── order.py
│ ├── value_objects/
│ │ ├── __init__.py
│ │ ├── email.py
│ │ └── money.py
│ ├── events/
│ │ ├── __init__.py
│ │ └── user_events.py
│ └── exceptions.py
│
├── application/ # 应用业务规则
│ ├── __init__.py
│ ├── use_cases/
│ │ ├── __init__.py
│ │ ├── create_user.py
│ │ └── place_order.py
│ ├── services/
│ │ ├── __init__.py
│ │ └── user_service.py
│ └── ports/
│ ├── __init__.py
│ ├── repositories.py
│ └── external_services.py
│
├── infrastructure/ # 外部接口
│ ├── __init__.py
│ ├── persistence/
│ │ ├── __init__.py
│ │ ├── sqlalchemy/
│ │ │ ├── models.py
│ │ │ └── repositories.py
│ │ └── mongodb/
│ │ └── repositories.py
│ ├── messaging/
│ │ ├── __init__.py
│ │ └── rabbitmq_publisher.py
│ ├── external_services/
│ │ ├── __init__.py
│ │ └── email_service.py
│ └── config.py
│
├── presentation/ # UI/API 层
│ ├── __init__.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── dependencies.py
│ │ ├── routes/
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── orders.py
│ │ └── schemas/
│ │ ├── __init__.py
│ │ └── user_schemas.py
│ └── cli/
│ └── commands.py
│
├── tests/
│ ├── unit/
│ ├── integration/
│ └── e2e/
│
├── main.py # 应用入口点
├── container.py # 依赖注入设置
├── pyproject.toml
└── README.md
测试整洁架构
测试领域逻辑的单元测试
import pytest
from uuid import uuid4
def test_user_creation():
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
assert user.email == "test@example.com"
assert user.is_active is True
def test_order_total_calculation():
order = Order(id=uuid4(), customer_id=uuid4())
order.add_item(
uuid4(),
quantity=2,
price=Money(10.0, "USD")
)
order.add_item(
uuid4(),
quantity=1,
price=Money(5.0, "USD")
)
assert order.total().amount == 25.0
与仓库集成测试
@pytest.fixture
def in_memory_repository():
return InMemoryUserRepository()
def test_user_repository_save_and_retrieve(in_memory_repository):
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
saved_user = in_memory_repository.save(user)
retrieved_user = in_memory_repository.get_by_id(user.id)
assert retrieved_user is not None
assert retrieved_user.email == user.email
服务层测试
from unittest.mock import Mock
def test_user_registration():
# 安排
mock_repository = Mock(spec=UserRepository)
mock_repository.get_by_email.return_value = None
mock_repository.save.return_value = User(
id=uuid4(),
email="test@example.com",
name="Test"
)
mock_email = Mock(spec=EmailSender)
mock_events = Mock(spec=EventPublisher)
service = UserService(mock_repository, mock_email, mock_events)
# 执行
user = service.register_user("test@example.com", "Test")
# 断言
assert user.email == "test@example.com"
mock_repository.save.assert_called_once()
mock_email.send.assert_called_once()
mock_events.publish.assert_called_once()
常见陷阱及避免方法
过度设计
不要为简单的 CRUD 应用程序实现整洁架构。从简单开始，并随着复杂性的增加进行重构。
泄漏的抽象
确保领域实体不包含数据库注解或特定于框架的代码：
# 不好
from sqlalchemy import Column
@dataclass
class User:
id: Column(Integer, primary_key=True) # 框架泄漏到领域
# 好
@dataclass
class User:
id: UUID # 纯领域对象
循环依赖
使用依赖注入和接口来打破各层之间的循环依赖。
忽略上下文
整洁架构并不是万能的。根据项目规模和团队专业知识调整各层的严格程度。
