名称: python-design-patterns 描述: Python设计模式包括KISS、关注点分离、单一职责和组合优于继承。在做出架构决策、重构代码结构或评估抽象适当性时使用。
Python设计模式
使用基本设计原则编写可维护的Python代码。这些模式帮助您构建易于理解、测试和修改的系统。
何时使用此技能
- 设计新组件或服务
- 重构复杂或纠缠的代码
- 决定是否创建抽象
- 在继承和组合之间选择
- 评估代码复杂性和耦合
- 规划模块化架构
核心概念
1. KISS (保持简单)
选择最简单的可行解决方案。复杂性必须由具体需求证明。
2. 单一职责 (SRP)
每个单元应该只有一个改变的理由。将关注点分离到专注的组件中。
3. 组合优于继承
通过组合对象而不是扩展类来构建行为。
4. 三的规则
在抽象之前等待直到有三个实例。重复通常比过早抽象更好。
快速开始
# 简单胜于聪明
# 而不是工厂/注册模式:
FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter}
def get_formatter(name: str) -> Formatter:
return FORMATTERS[name]()
基本模式
模式 1: KISS - 保持简单
在添加复杂性之前,问:更简单的解决方案是否可行?
# 过度设计:带注册的工厂
class OutputFormatterFactory:
_formatters: dict[str, type[Formatter]] = {}
@classmethod
def register(cls, name: str):
def decorator(formatter_cls):
cls._formatters[name] = formatter_cls
return formatter_cls
return decorator
@classmethod
def create(cls, name: str) -> Formatter:
return cls._formatters[name]()
@OutputFormatterFactory.register("json")
class JsonFormatter(Formatter):
...
# 简单:只使用字典
FORMATTERS = {
"json": JsonFormatter,
"csv": CsvFormatter,
"xml": XmlFormatter,
}
def get_formatter(name: str) -> Formatter:
"""通过名称获取格式化器。"""
if name not in FORMATTERS:
raise ValueError(f"未知格式: {name}")
return FORMATTERS[name]()
工厂模式在这里添加代码而没有添加价值。将模式留到解决实际问题时使用。
模式 2: 单一职责原则
每个类或函数应该只有一个改变的理由。
# 不好:处理程序做所有事情
class UserHandler:
async def create_user(self, request: Request) -> Response:
# HTTP解析
data = await request.json()
# 验证
if not data.get("email"):
return Response({"error": "需要邮箱"}, status=400)
# 数据库访问
user = await db.execute(
"INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *",
data["email"], data["name"]
)
# 响应格式化
return Response({"id": user.id, "email": user.email}, status=201)
# 好:分离关注点
class UserService:
"""仅业务逻辑。"""
def __init__(self, repo: UserRepository) -> None:
self._repo = repo
async def create_user(self, data: CreateUserInput) -> User:
# 仅业务规则
user = User(email=data.email, name=data.name)
return await self._repo.save(user)
class UserHandler:
"""仅HTTP关注点。"""
def __init__(self, service: UserService) -> None:
self._service = service
async def create_user(self, request: Request) -> Response:
data = CreateUserInput(**(await request.json()))
user = await self._service.create_user(data)
return Response(user.to_dict(), status=201)
现在HTTP更改不影响业务逻辑,反之亦然。
模式 3: 关注点分离
将代码组织成具有明确职责的独立层。
┌─────────────────────────────────────────────────────┐
│ API层 (处理程序) │
│ - 解析请求 │
│ - 调用服务 │
│ - 格式化响应 │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 服务层 (业务逻辑) │
│ - 领域规则和验证 │
│ - 协调操作 │
│ - 尽可能使用纯函数 │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 仓库层 (数据访问) │
│ - SQL查询 │
│ - 外部API调用 │
│ - 缓存操作 │
└─────────────────────────────────────────────────────┘
每层仅依赖于其下的层:
# 仓库:数据访问
class UserRepository:
async def get_by_id(self, user_id: str) -> User | None:
row = await self._db.fetchrow(
"SELECT * FROM users WHERE id = $1", user_id
)
return User(**row) if row else None
# 服务:业务逻辑
class UserService:
def __init__(self, repo: UserRepository) -> None:
self._repo = repo
async def get_user(self, user_id: str) -> User:
user = await self._repo.get_by_id(user_id)
if user is None:
raise UserNotFoundError(user_id)
return user
# 处理程序:HTTP关注点
@app.get("/users/{user_id}")
async def get_user(user_id: str) -> UserResponse:
user = await user_service.get_user(user_id)
return UserResponse.from_user(user)
模式 4: 组合优于继承
通过组合对象而不是继承来构建行为。
# 继承:僵化且难以测试
class EmailNotificationService(NotificationService):
def __init__(self):
super().__init__()
self._smtp = SmtpClient() # 难以模拟
def notify(self, user: User, message: str) -> None:
self._smtp.send(user.email, message)
# 组合:灵活且可测试
class NotificationService:
"""通过多个渠道发送通知。"""
def __init__(
self,
email_sender: EmailSender,
sms_sender: SmsSender | None = None,
push_sender: PushSender | None = None,
) -> None:
self._email = email_sender
self._sms = sms_sender
self._push = push_sender
async def notify(
self,
user: User,
message: str,
channels: set[str] | None = None,
) -> None:
channels = channels or {"email"}
if "email" in channels:
await self._email.send(user.email, message)
if "sms" in channels and self._sms and user.phone:
await self._sms.send(user.phone, message)
if "push" in channels and self._push and user.device_token:
await self._push.send(user.device_token, message)
# 易于用假对象测试
service = NotificationService(
email_sender=FakeEmailSender(),
sms_sender=FakeSmsSender(),
)
高级模式
模式 5: 三的规则
在抽象之前等待直到有三个实例。
# 两个相似函数?先别抽象
def process_orders(orders: list[Order]) -> list[Result]:
results = []
for order in orders:
validated = validate_order(order)
result = process_validated_order(validated)
results.append(result)
return results
def process_returns(returns: list[Return]) -> list[Result]:
results = []
for ret in returns:
validated = validate_return(ret)
result = process_validated_return(validated)
results.append(result)
return results
# 这些看起来相似,但等等!它们实际上相同吗?
# 不同的验证、不同的处理、不同的错误...
# 重复通常比错误的抽象更好
# 只有在第三个案例后,才考虑是否有真正的模式
# 但即使那时,有时明确比抽象更好
模式 6: 函数大小指导
保持函数专注。在函数符合以下情况时提取:
- 超过20-50行(根据复杂性变化)
- 服务于多个不同目的
- 有深度嵌套的逻辑(3+级)
# 太长,多个关注点混合
def process_order(order: Order) -> Result:
# 50行验证...
# 30行库存检查...
# 40行支付处理...
# 20行通知...
pass
# 更好:由专注的函数组成
def process_order(order: Order) -> Result:
"""通过完整工作流程处理客户订单。"""
validate_order(order)
reserve_inventory(order)
payment_result = charge_payment(order)
send_confirmation(order, payment_result)
return Result(success=True, order_id=order.id)
模式 7: 依赖注入
通过构造函数传递依赖以实现可测试性。
from typing import Protocol
class Logger(Protocol):
def info(self, msg: str, **kwargs) -> None: ...
def error(self, msg: str, **kwargs) -> None: ...
class Cache(Protocol):
async def get(self, key: str) -> str | None: ...
async def set(self, key: str, value: str, ttl: int) -> None: ...
class UserService:
"""带注入依赖的服务。"""
def __init__(
self,
repository: UserRepository,
cache: Cache,
logger: Logger,
) -> None:
self._repo = repository
self._cache = cache
self._logger = logger
async def get_user(self, user_id: str) -> User:
# 先检查缓存
cached = await self._cache.get(f"user:{user_id}")
if cached:
self._logger.info("缓存命中", user_id=user_id)
return User.from_json(cached)
# 从数据库获取
user = await self._repo.get_by_id(user_id)
if user:
await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300)
return user
# 生产环境
service = UserService(
repository=PostgresUserRepository(db),
cache=RedisCache(redis),
logger=StructlogLogger(),
)
# 测试环境
service = UserService(
repository=InMemoryUserRepository(),
cache=FakeCache(),
logger=NullLogger(),
)
模式 8: 避免常见反模式
不要暴露内部类型:
# 不好:泄露ORM模型到API
@app.get("/users/{id}")
def get_user(id: str) -> UserModel: # SQLAlchemy模型
return db.query(UserModel).get(id)
# 好:使用响应模式
@app.get("/users/{id}")
def get_user(id: str) -> UserResponse:
user = db.query(UserModel).get(id)
return UserResponse.from_orm(user)
不要混合I/O与业务逻辑:
# 不好:SQL嵌入业务逻辑
def calculate_discount(user_id: str) -> float:
user = db.query("SELECT * FROM users WHERE id = ?", user_id)
orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
# 业务逻辑与数据访问混合
# 好:仓库模式
def calculate_discount(user: User, order_history: list[Order]) -> float:
# 纯业务逻辑,易于测试
if len(order_history) > 10:
return 0.15
return 0.0
最佳实践总结
- 保持简单 - 选择最简单的可行解决方案
- 单一职责 - 每个单元有一个改变的理由
- 分离关注点 - 具有明确目的的不同层
- 组合优于继承 - 组合对象以获得灵活性
- 三的规则 - 在抽象之前等待
- 保持函数小 - 20-50行(根据复杂性变化),一个目的
- 注入依赖 - 构造函数注入以实现可测试性
- 在抽象之前删除 - 移除死代码,然后考虑模式
- 测试每层 - 每个关注点的隔离测试
- 明确胜于聪明 - 可读的代码胜过优雅的代码