8.2 可维护性与SOLID：降低修改风险
可维护性衡量了软件在需求变更、缺陷修复和性能优化时，能够多大程度上被容易地修改。遵循SOLID原则的代码，具有以下显著的可维护性优势：
第八章：SOLID原则在测试与维护中的高级策略 (续)
…遵循SOLID原则的代码，具有以下显著的可维护性优势：
局部化修改影响 (OCP, SRP)：
当需求变更需要修改代码时，遵循SRP的类意味着修改通常只集中在少数几个类或模块中，因为每个类只负责一个单一的职责。
遵循OCP则进一步确保了这些修改不会波及到系统的其他部分。通过依赖抽象和提供扩展点，新功能或变更可以作为增量添加，而不需要触碰已稳定和测试过的核心代码。这种局部化的修改能力，极大地降低了引入新Bug的风险。
例如，如果我们需要改变用户验证规则，我们只需要修改 UserValidator 类；如果我们需要更换数据库类型，我们只需要修改 UserRepository 的具体实现；这些修改不会影响到使用这些组件的 UserService 或 AuthController。
简化缺陷修复 (SRP)：
当出现Bug时，由于每个组件的职责单一且明确，Bug的根源更容易被定位。如果一个错误发生在邮件发送过程中，我们知道应该去检查 EmailService 及其相关配置，而不是在整个项目中大海捞针。
隔离的职责也意味着修复Bug时，对代码的修改范围更小，更容易进行回归测试，确保修复本身没有引入新的问题。
提高代码可理解性 (SRP, ISP)：
职责单一的类（SRP）和接口隔离的模块（ISP）使得代码更容易被阅读和理解。当一个新开发者加入团队时，他们可以更快地理解每个组件的作用和行为。
小而精的接口减少了理解一个模块所需的信息量。你不需要了解一个接口的所有方法，只需要关心你实际使用的那些。这降低了认知负担。
降低回归风险 (DIP, LSP)：
通过依赖抽象（DIP），高层模块与低层实现解耦。这意味着你可以更换底层实现，只要它们遵循相同的抽象契约（LSP），高层模块就不需要修改，并且其原有功能也不太可能受到影响。
当进行重构或升级底层库时，这种解耦提供了强大的保障。你可以独立地替换和测试组件，而不必担心意外地破坏系统的其他部分。
简化新成员上手：
一个结构良好、遵循SOLID原则的代码库，其模块边界清晰，依赖关系明确，有助于新入职的开发者更快地理解系统架构和业务逻辑。他们可以更容易地找到需要修改的代码位置，并自信地进行更改，因为他们知道他们的修改不太可能产生广泛的副作用。
可维护性示例：策略模式与OCP
假设我们有一个订单支付系统，支持多种支付方式：信用卡、支付宝、微信支付。
维护性差的设计（违反 OCP）：
class BadPaymentProcessor: # 糟糕的支付处理器
def process_payment(self, payment_type: str, amount: float, details: dict): # 处理支付方法
if payment_type == "credit_card": # 如果支付类型是信用卡
print(f"处理信用卡支付: 金额 {
amount}, 详情 {
details}") # 打印信息
# ... 复杂的信用卡支付API调用逻辑 ...
if amount > 1000: # 模拟金额过大
raise ValueError("信用卡支付金额过大。") # 抛出值错误
print("信用卡支付成功。") # 打印成功信息
return {
"status": "success", "transaction_id": "CC123"} # 返回成功信息
elif payment_type == "alipay": # 如果支付类型是支付宝
print(f"处理支付宝支付: 金额 {
amount}, 详情 {
details}") # 打印信息
# ... 复杂的支付宝支付API调用逻辑 ...
print("支付宝支付成功。") # 打印成功信息
return {
"status": "success", "transaction_id": "ALIPAY456"} # 返回成功信息
elif payment_type == "wechat_pay": # 如果支付类型是微信支付
print(f"处理微信支付: 金额 {
amount}, 详情 {
details}") # 打印信息
# ... 复杂的微信支付API调用逻辑 ...
print("微信支付成功。") # 打印成功信息
return {
"status": "success", "transaction_id": "WECHAT789"} # 返回成功信息
else: # 其他支付类型
raise ValueError(f"不支持的支付类型: {
payment_type}") # 抛出值错误
# 客户端代码
# processor = BadPaymentProcessor()
# processor.process_payment("credit_card", 500, {"card_num": "1234..."})
# processor.process_payment("alipay", 200, {"order_info": "..."})
#
# # 问题：如果新增一种支付方式 (如 PayPal)，需要修改 process_payment 方法，违反 OCP。
# # 每次修改都会增加出错的风险，并需要重新测试所有支付方式。
可维护性好的设计（遵循 OCP, DIP, SRP – 使用策略模式）：
import abc # 导入 abc 模块
# 1. 抽象支付策略 (DIP)
class PaymentStrategy(abc.ABC): # 支付策略抽象基类
@abc.abstractmethod # 抽象方法
def pay(self, amount: float, details: dict) -> dict: # 支付方法
pass
# 2. 具体支付策略 (SRP)
class CreditCardPayment(PaymentStrategy): # 信用卡支付类
def pay(self, amount: float, details: dict) -> dict: # 实现支付方法
print(f"CreditCardPayment: 处理信用卡支付: 金额 {
amount}, 详情 {
details}") # 打印信息
# ... 实际的信用卡支付API调用逻辑 ...
if amount > 1000: # 模拟业务规则
raise ValueError("信用卡单笔支付金额不能超过1000。") # 抛出值错误
print("CreditCardPayment: 信用卡支付成功。") # 打印成功信息
return {
"status": "success", "transaction_id": f"CC_{
time.time_ns()}"} # 返回成功信息
class AlipayPayment(PaymentStrategy): # 支付宝支付类
def pay(self, amount: float, details: dict) -> dict: # 实现支付方法
print(f"AlipayPayment: 处理支付宝支付: 金额 {
amount}, 详情 {
details}") # 打印信息
# ... 实际的支付宝支付API调用逻辑 ...
print("AlipayPayment: 支付宝支付成功。") # 打印成功信息
return {
"status": "success", "transaction_id": f"ALIPAY_{
time.time_ns()}"} # 返回成功信息
class WeChatPayPayment(PaymentStrategy): # 微信支付类
def pay(self, amount: float, details: dict) -> dict: # 实现支付方法
print(f"WeChatPayPayment: 处理微信支付: 金额 {
amount}, 详情 {
details}") # 打印信息
# ... 实际的微信支付API调用逻辑 ...
print("WeChatPayPayment: 微信支付成功。") # 打印成功信息
return {
"status": "success", "transaction_id": f"WECHAT_{
time.time_ns()}"} # 返回成功信息
# 3. 支付上下文/处理器 (OCP, DIP)
class GoodPaymentProcessor: # 好的支付处理器
def __init__(self, strategy: PaymentStrategy): # 构造函数
self._strategy = strategy # 注入支付策略
def process_payment(self, amount: float, details: dict) -> dict: # 处理支付方法
print(f"GoodPaymentProcessor: 正在通过策略处理支付...") # 打印信息
return self._strategy.pay(amount, details) # 调用策略的支付方法
# 4. 支付策略工厂 (OCP, SRP)
class PaymentStrategyFactory: # 支付策略工厂
def __init__(self): # 构造函数
self._strategies = {
# 策略字典
"credit_card": CreditCardPayment(), # 信用卡支付
"alipay": AlipayPayment(), # 支付宝支付
"wechat_pay": WeChatPayPayment() # 微信支付
} # 策略字典
def get_strategy(self, payment_type: str) -> PaymentStrategy: # 获取策略
strategy = self._strategies.get(payment_type) # 获取策略
if not strategy: # 如果策略不存在
raise ValueError(f"不支持的支付类型: {
payment_type}") # 抛出值错误
return strategy # 返回策略
# 客户端代码
print("--- 策略模式实现可维护的支付系统 ---")
factory = PaymentStrategyFactory() # 创建策略工厂
# 信用卡支付
print("
=== 信用卡支付 ===")
credit_card_strategy = factory.get_strategy("credit_card") # 获取信用卡策略
credit_card_processor = GoodPaymentProcessor(credit_card_strategy) # 创建处理器
credit_card_processor.process_payment(750.0, {
"card_number": "1234-5678-9012-3456"}) # 处理支付
try: # 尝试大额支付
credit_card_processor.process_payment(1200.0, {
"card_number": "9876..."}) # 尝试大额支付
except ValueError as e: # 捕获值错误
print(f"客户端捕获到错误 (预期): {
e}") # 打印错误信息
# 支付宝支付
print("
=== 支付宝支付 ===")
alipay_strategy = factory.get_strategy("alipay") # 获取支付宝策略
alipay_processor = GoodPaymentProcessor(alipay_strategy) # 创建处理器
alipay_processor.process_payment(200.0, {
"buyer_id": "user123"}) # 处理支付
# 扩展性演示：新增 PayPal 支付 (无需修改 GoodPaymentProcessor 或 PaymentStrategyFactory.get_strategy)
class PayPalPayment(PaymentStrategy): # PayPal 支付类
def pay(self, amount: float, details: dict) -> dict: # 实现支付方法
print(f"PayPalPayment: 处理PayPal支付: 金额 {
amount}, 详情 {
details}") # 打印信息
# ... 实际的PayPal支付API调用逻辑 ...
print("PayPalPayment: PayPal支付成功。") # 打印成功信息
return {
"status": "success", "transaction_id": f"PAYPAL_{
time.time_ns()}"} # 返回成功信息
print("
--- 扩展性：新增 PayPal 支付 (OCP) ---")
# 只需在工厂注册新策略，无需修改现有代码
factory._strategies["paypal"] = PayPalPayment() # 直接添加到工厂的内部字典 (简化演示，实际可能通过配置加载)
paypal_strategy = factory.get_strategy("paypal") # 获取 PayPal 策略
paypal_processor = GoodPaymentProcessor(paypal_strategy) # 创建处理器
paypal_processor.process_payment(300.0, {
"paypal_email": "test@example.com"}) # 处理支付
这个重构后的支付系统，通过策略模式完美体现了SOLID原则，从而极大地提升了可维护性：
SRP： 每个支付类 (CreditCardPayment, AlipayPayment, WeChatPayPayment, PayPalPayment) 只专注于处理一种支付方式的细节。GoodPaymentProcessor 只负责接收支付请求并将其委托给相应的策略。PaymentStrategyFactory 的职责是根据类型提供正确的策略实例。
OCP： 如果需要添加新的支付方式，只需创建一个实现 PaymentStrategy 接口的新类，并将其注册到 PaymentStrategyFactory 中即可。GoodPaymentProcessor 的核心逻辑（process_payment 方法）无需修改。这种对扩展开放、对修改封闭的特性，使得系统在维护和迭代时更加稳定，降低了引入Bug的风险。
DIP： GoodPaymentProcessor（高层模块）依赖于 PaymentStrategy（抽象接口），而不是具体的支付实现。客户端代码通过 PaymentStrategyFactory 间接获取策略，进一步解耦。
LSP： 任何 PaymentStrategy 的子类都能够替换其他的支付策略，并且 GoodPaymentProcessor 能够正确地调用它们的 pay 方法，而不会引起意外行为。
8.3 SOLID原则与技术债 (Technical Debt) 的管理
技术债 (Technical Debt) 是指为了加快开发速度或简化短期工作而选择的非最优解决方案，这些“捷径”会导致未来的开发、维护和扩展成本增加。就像财务上的债务一样，技术债需要定期“偿还”，否则它会随着时间积累利息，使项目变得越来越难以管理。
8.3.1 违反SOLID如何积累技术债
违反SRP（职责膨胀）： 当一个类或函数承担了过多职责时，它会变得庞大而复杂，难以理解和测试。每次修改都会触及不相关的逻辑，导致潜在的Bug。这种“牵一发而动全身”的现象使得修改成本和风险急剧增加。
例子： 一个 UserManager 类既处理用户注册、登录，又发送邮件、管理数据库连接、生成用户报告。
违反OCP（僵硬的代码）： 当系统对扩展不开放、对修改不封闭时，每次添加新功能都意味着要修改现有已测试通过的代码，导致“打补丁”式的开发。这种代码很“僵硬”，难以适应变化。
例子： 支付处理器中大量的 if/elif 链，每次添加新支付方式都需要修改核心方法。
违反DIP（紧耦合）： 当高层模块直接依赖低层具体实现时，它们之间会形成紧密的耦合。底层实现的任何变化都会向上层传递，导致连锁反应式的修改。这种代码很“脆弱”，一处改动可能导致多处崩溃。
例子： 业务逻辑层直接调用特定数据库连接对象的 execute_sql_query 方法。
违反LSP（破碎的继承）： 当子类不能安全地替换父类时，继承关系就会变得混乱和危险。调用者在使用多态时必须进行类型检查，或者面对意料之外的行为，增加了Bug和测试的复杂性。
例子： 一个 Rectangle 和 Square 的继承关系，其中 Square 覆盖了 setWidth 和 setHeight 导致非正方形行为。
违反ISP（胖接口）： 庞大的接口使得实现者需要实现它不需要的方法，或者客户端被迫依赖它不关心的方法。这增加了模块间的耦合，降低了灵活性。
例子： 一个 IDataStore 接口包含了文件操作、网络操作和数据库操作的所有方法。
这些违反SOLID原则的设计模式，都会导致代码质量下降，模块间耦合度提高，可扩展性和可维护性降低，最终积累成难以偿还的技术债。
8.3.2 SOLID如何帮助管理技术债
预防技术债的产生： 从一开始就遵循SOLID原则进行设计，可以帮助你构建出模块化、低耦合、高内聚的代码库，从而从源头上减少技术债的产生。它鼓励开发者在编写代码时就考虑未来的变化和扩展。
识别和量化技术债： 当你发现一个类或方法难以测试、修改时，或者每次添加功能都像在打补丁时，这通常是违反SOLID原则的信号，也是技术债存在的明显标志。SOLID原则提供了一种诊断技术债的框架。
指导技术债的偿还： 当你决定重构以偿还技术债时，SOLID原则提供了清晰的重构方向和目标。
职责拆分（SRP）： 将臃肿的类或函数拆分成更小、更专注的单元。
引入抽象（DIP）： 解耦紧密耦合的模块，使得它们可以独立演进。
策略模式、工厂模式（OCP）： 消除 if/elif 链，使得系统对扩展开放。
细化接口（ISP）： 减少不必要的依赖。
检查继承关系（LSP）： 确保多态的正确性。
示例：重构违反SRP的技术债
假设我们有一个遗留的 UserProcessor 类，它负责用户的所有操作，包括验证、数据库操作和通知。
初始技术债代码 (违反 SRP)：
class LegacyUserProcessor: # 遗留用户处理器
def __init__(self, db_connection_string: str): # 构造函数
self.db_conn_str = db_connection_string # 数据库连接字符串
print(f"LegacyUserProcessor: 使用数据库连接字符串 '{
self.db_conn_str}'。") # 打印信息
def create_user(self, username: str, email: str, password: str) -> dict: # 创建用户
print(f"
LegacyUserProcessor: 尝试创建用户 {
username}。") # 打印信息
# 1. 输入验证 (职责A)
if not username or not email or not password: # 如果用户名、邮箱或密码为空
return {
"status": "error", "message": "所有字段都必须填写。"} # 返回错误信息
if "@" not in email: # 如果邮箱不包含 @
return {
"status": "error", "message": "邮箱格式不正确。"} # 返回错误信息
# 2. 数据库操作 (职责B)
# 模拟数据库检查用户名和邮箱是否存在
if username == "admin_legacy" or email == "admin@legacy.com": # 如果用户名或邮箱已存在
return {
"status": "error", "message": "用户名或邮箱已存在。"} # 返回错误信息
# 模拟保存到数据库
new_user_id = f"legacy_{
username}_id" # 生成模拟用户ID
print(f"LegacyUserProcessor: 用户 {
username} 已保存到数据库。") # 打印信息
# 3. 发送通知 (职责C)
if email: # 如果有邮箱
print(f"LegacyUserProcessor: 发送欢迎邮件到 {
email}。") # 打印信息
# ... 邮件发送逻辑 ...
return {
"status": "success", "user_id": new_user_id, "message": "用户创建成功。"} # 返回成功信息
# 客户端调用
# processor = LegacyUserProcessor("sql_db_prod")
# processor.create_user("john_doe", "john@example.com", "pass123")
# processor.create_user("", "invalid@example.com", "pass") # 验证失败
偿还技术债：遵循SRP的重构
import abc
# 1. 用户验证器 (SRP)
class UserValidatorRefactored(abc.ABC): # 用户验证器重构版抽象基类
@abc.abstractmethod # 抽象方法
def validate_user_data(self, data: dict) -> tuple[bool, str]: # 验证用户数据
pass
class BasicUserValidator(UserValidatorRefactored): # 基本用户验证器
def validate_user_data(self, data: dict) -> tuple[bool, str]: # 实现验证用户数据
username = data.get("username") # 获取用户名
email = data.get("email") # 获取邮箱
password = data.get("password") # 获取密码
if not username or not email or not password: # 如果用户名、邮箱或密码为空
return False, "所有字段都必须填写。" # 返回失败和错误信息
if "@" not in email: # 如果邮箱不包含 @
return False, "邮箱格式不正确。" # 返回失败和错误信息
return True, "验证通过。" # 返回成功和信息
# 2. 用户数据仓库 (SRP, DIP)
class UserRepositoryRefactored(abc.ABC): # 用户数据仓库重构版抽象基类
@abc.abstractmethod # 抽象方法
def get_by_username(self, username: str) -> dict: # 根据用户名获取用户
pass
@abc.abstractmethod # 抽象方法
def get_by_email(self, email: str) -> dict: # 根据邮箱获取用户
pass
@abc.abstractmethod # 抽象方法
def create_user(self, user_data: dict) -> dict: # 创建用户
pass
class InMemoryUserRepositoryRefactored(UserRepositoryRefactored): # 内存用户数据仓库重构版
def __init__(self): # 构造函数
self._users = {
} # 模拟内存数据库
self._next_id = 1 # 下一个 ID
def get_by_username(self, username: str) -> dict: # 实现根据用户名获取用户
for user_id, user in self._users.items(): # 遍历用户
if user["username"] == username: # 如果用户名匹配
return user # 返回用户
return None # 返回 None
def get_by_email(self, email: str) -> dict: # 实现根据邮箱获取用户
for user_id, user in self._users.items(): # 遍历用户
if user["email"] == email: # 如果邮箱匹配
return user # 返回用户
return None # 返回 None
def create_user(self, user_data: dict) -> dict: # 实现创建用户
user_id = f"user_{
self._next_id}" # 生成用户 ID
self._next_id += 1 # 递增 ID
new_user = user_data.copy() # 复制用户数据
new_user["user_id"] = user_id # 添加用户 ID
self._users[user_id] = new_user # 保存用户
return new_user # 返回新用户
# 3. 邮件通知服务 (SRP, DIP)
class EmailServiceRefactored(abc.ABC): # 邮件服务重构版抽象基类
@abc.abstractmethod # 抽象方法
def send_welcome_email(self, recipient_email: str, username: str): # 发送欢迎邮件
pass
class MockEmailServiceRefactored(EmailServiceRefactored): # 模拟邮件服务重构版
def send_welcome_email(self, recipient_email: str, username: str): # 实现发送欢迎邮件
print(f"MockEmailServiceRefactored: 模拟发送欢迎邮件到 {
recipient_email} 给 {
username}。") # 打印信息
# 4. 核心用户业务服务 (SRP, 协调职责，DIP 依赖抽象)
class UserServiceRefactored: # 用户服务重构版
def __init__(self, validator: UserValidatorRefactored, # 注入验证器
repository: UserRepositoryRefactored, # 注入数据仓库
email_service: EmailServiceRefactored): # 注入邮件服务
self._validator = validator # 验证器
self._repository = repository # 数据仓库
self._email_service = email_service # 邮件服务
def register_user(self, user_data: dict) -> dict: # 注册用户方法
# 1. 验证输入
is_valid, msg = self._validator.validate_user_data(user_data) # 验证用户数据
if not is_valid: # 如果验证失败
raise ValueError(f"输入验证失败: {
msg}") # 抛出值错误
# 2. 检查业务规则 (如唯一性)
if self._repository.get_by_username(user_data["username"]): # 如果用户名已存在
raise ValueError("用户名已存在。") # 抛出值错误
if self._repository.get_by_email(user_data["email"]): # 如果邮箱已存在
raise ValueError("邮箱已存在。") # 抛出值错误
# 3. 创建用户
new_user = self._repository.create_user(user_data) # 创建用户
# 4. 发送通知
self._email_service.send_welcome_email(new_user["email"], new_user["username"]) # 发送欢迎邮件
print(f"UserServiceRefactored: 用户 {
new_user['username']} 注册成功。") # 打印信息
return {
"status": "success", "user_id": new_user["user_id"]} # 返回成功信息
# 客户端代码 (组装和使用重构后的服务)
print("
--- 偿还技术债：SOLID重构后的用户注册 ---")
validator = BasicUserValidator() # 创建基本用户验证器
repository = InMemoryUserRepositoryRefactored() # 创建内存用户数据仓库
email_service = MockEmailServiceRefactored() # 创建模拟邮件服务
user_service = UserServiceRefactored(validator, repository, email_service) # 创建用户服务
# 成功注册
try: # 尝试注册用户
user_service.register_user({
"username": "new_user_solid", "email": "solid@example.com", "password": "pass"}) # 注册用户
except ValueError as e: # 捕获值错误
print(f"注册失败 (不应该发生): {
e}") # 打印错误信息
# 验证失败
try: # 尝试注册无效用户
user_service.register_user({
"username": "fail_user", "email": "invalid_email", "password": "pass"}) # 注册无效用户
except ValueError as e: # 捕获值错误
print(f"注册失败 (预期): {
e}") # 打印错误信息
# 用户名已存在
try: # 尝试注册重复用户名
user_service.register_user({
"username": "new_user_solid", "email": "another@example.com", "password": "pass"}) # 注册重复用户名
except ValueError as e: # 捕获值错误
print(f"注册失败 (预期): {
e}") # 打印错误信息
通过这次重构，我们成功地将 LegacyUserProcessor 的职责拆分到了 UserValidatorRefactored、UserRepositoryRefactored 和 EmailServiceRefactored 中，并由 UserServiceRefactored 来协调。
SRP： 每个类现在都有了单一的职责，它们只因一个原因而改变。
DIP： UserServiceRefactored 依赖于抽象接口，而非具体实现。
OCP： 如果验证规则、数据存储方式或邮件服务提供商发生变化，只需修改相应的具体实现类，而 UserServiceRefactored 无需修改。
这不仅使得代码更易于理解和测试，还大大降低了未来的修改风险，有效地“偿还”了原有的技术债。
8.4 架构演进与SOLID原则：应对未来变化
软件系统很少是一成不变的。业务需求会不断演进，技术栈会更新换代，用户量和性能要求也会不断增长。一个好的架构应该能够支持这种演进 (Evolutionary Architecture)，而不是成为变化的阻碍。SOLID原则是实现这一目标的关键基石。
8.4.1 拥抱不确定性与弹性设计
识别变化点： SOLID原则通过鼓励职责分离和依赖抽象，帮助我们识别系统中可能发生变化的“点”。这些变化点（如数据存储、外部服务、业务规则的变体）正是我们可以应用OCP和DIP来构建弹性设计的区域。
降低变化成本： 当变化发生时，如果系统遵循SOLID，那么所需的修改将是局部化的，并且可以通过扩展而非修改现有代码来完成。这大大降低了每次变更的成本、风险和所需的时间。
促进小步快跑： 遵循SOLID原则的模块化设计，使得团队可以进行小规模、增量的变更和部署，这与敏捷开发和持续交付的理念高度契合。
8.4.2 在遗留系统中渐进式采用SOLID
对于已经存在的大型遗留系统，全面遵循SOLID原则可能不切实际，甚至会带来更大的风险。在这种情况下，应采取渐进式重构策略：
识别痛点： 找出系统中修改最频繁、最容易出错、最难以测试的“坏味道”区域。这些通常是违反SOLID原则最严重的地方。
隔离边界： 尝试通过引入抽象或适配器，逐步将这些痛点模块与其外部依赖解耦。这通常是引入DIP的第一步。
小步重构： 每次只针对一个小的、明确的职责进行拆分（SRP），或者引入一个策略来消除 if/elif 链（OCP）。
编写测试： 在重构之前，为需要修改的遗留代码编写回归测试，以确保重构过程不会破坏现有功能。重构后的代码也应该有良好的单元测试覆盖。
增量交付： 将重构后的代码作为小的、可部署的单元逐渐集成到主分支中。
优先重构新功能： 在开发新功能时，强制要求新代码遵循SOLID原则。这有助于逐渐将新架构的健康部分扩展到整个系统。
修缮模式 (Strangler Fig Pattern)： 对于特别庞大和复杂的遗留模块，可以采用修缮模式。逐步用新的、遵循SOLID原则的组件来替代旧系统的功能，就像绞杀榕树逐渐覆盖并取代宿主树一样。
示例：使用适配器模式处理遗留API（DIP，OCP）
假设我们有一个遗留的 LegacyPaymentGateway，它暴露了一个不友好的、紧密耦合的API。我们希望在新的支付系统中遵循SOLID原则，同时仍然能够调用这个遗留网关。
遗留 API (假定无法修改)：
class LegacyPaymentGateway: # 遗留支付网关
def init_transaction(self, raw_data: dict) -> dict: # 初始化事务
print(f"LegacyGateway: 初始化事务，数据: {
raw_data}") # 打印信息
if not raw_data.get("amount") or raw_data["amount"] <= 0: # 如果金额为空或小于等于0
raise ValueError("Legacy: 金额无效。") # 抛出值错误
return {
"legacy_txn_id": "LGC" + str(time.time_ns()), "status": "PENDING_LEGACY"} # 返回事务ID和状态
def execute_payment(self, legacy_txn_id: str, card_info: dict) -> dict: # 执行支付
print(f"LegacyGateway: 执行支付 for {
legacy_txn_id}，卡信息: {
card_info.get('number', '***')}") # 打印信息
if len(card_info.get("number", "")) < 16: # 如果卡号小于16位
raise ValueError("Legacy: 无效卡号。") # 抛出值错误
return {
"status": "COMPLETED_LEGACY", "legacy_message": "Legacy payment processed."} # 返回状态和消息
def query_status(self, legacy_txn_id: str) -> dict: # 查询状态
print(f"LegacyGateway: 查询状态 for {
legacy_txn_id}") # 打印信息
# 模拟不同的状态
if "LGC" in legacy_txn_id: # 如果包含 LGC
return {
"status": "COMPLETED_LEGACY", "details": "Query successful."} # 返回完成状态
return {
"status": "FAILED_LEGACY"} # 返回失败状态
引入适配器以遵循SOLID (DIP, OCP, SRP)：
import abc # 导入 abc 模块
# 1. 新支付系统定义的抽象接口 (DIP 的目标)
class NewPaymentGateway(abc.ABC): # 新支付网关抽象基类
@abc.abstractmethod # 抽象方法
def process_payment(self, amount: float, card_number: str, expiry_date: str) -> dict: # 处理支付方法
pass
@abc.abstractmethod # 抽象方法
def get_transaction_status(self, transaction_id: str) -> str: # 获取事务状态
pass
# 2. 适配器 (SRP: 职责是适配遗留API到新接口)
class LegacyGatewayAdapter(NewPaymentGateway): # 遗留网关适配器
def __init__(self, legacy_gateway: LegacyPaymentGateway): # 构造函数
self._legacy_gateway = legacy_gateway # 注入遗留网关
def process_payment(self, amount: float, card_number: str, expiry_date: str) -> dict: # 实现处理支付
print("
Adapter: 适配 LegacyPaymentGateway 的 process_payment。") # 打印信息
# 将新接口的数据模型转换为遗留API所需的数据模型
init_data = {
"amount": amount, "currency": "USD"} # 初始化数据
try: # 尝试初始化事务
init_response = self._legacy_gateway.init_transaction(init_data) # 初始化事务
legacy_txn_id = init_response["legacy_txn_id"] # 获取遗留事务ID
card_info = {
"number": card_number, "expiry": expiry_date} # 卡信息
exec_response = self._legacy_gateway.execute_payment(legacy_txn_id, card_info) # 执行支付
# 将遗留API的响应转换为新接口的统一响应
if exec_response["status"] == "COMPLETED_LEGACY": # 如果状态是已完成
return {
"status": "SUCCESS", "new_txn_id": legacy_txn_id, "message": "Payment processed by legacy gateway."} # 返回成功
else: # 否则
return {
"status": "FAILED", "new_txn_id": legacy_txn_id, "message": exec_response.get("legacy_message", "Unknown legacy error.")} # 返回失败
except ValueError as e: # 捕获值错误
return {
"status": "ERROR", "message": f"Adapter Error: {
e}"} # 返回错误
def get_transaction_status(self, transaction_id: str) -> str: # 实现获取事务状态
print(f"Adapter: 适配 LegacyPaymentGateway 的 get_transaction_status for {
transaction_id}。") # 打印信息
status_response = self._legacy_gateway.query_status(transaction_id) # 查询状态
# 将遗留状态映射到新系统的统一状态
if status_response["status"] == "COMPLETED_LEGACY": # 如果状态是已完成
return "COMPLETED" # 返回 COMPLETED
elif status_response["status"] == "PENDING_LEGACY": # 如果状态是待处理
return "PENDING" # 返回 PENDING
return "UNKNOWN" # 返回 UNKNOWN
# 3. 新支付服务 (DIP: 依赖抽象 NewPaymentGateway)
class NewPaymentService: # 新支付服务
def __init__(self, payment_gateway: NewPaymentGateway): # 构造函数
self._payment_gateway = payment_gateway # 注入支付网关
def charge_customer(self, amount: float, card_num: str, expiry: str) -> dict: # 客户扣款
print(f"NewPaymentService: 正在尝试扣款 {
amount}。") # 打印信息
return self._payment_gateway.process_payment(amount, card_num, expiry) # 调用支付网关处理支付
def check_payment_status(self, txn_id: str) -> str: # 检查支付状态
print(f"NewPaymentService: 正在检查事务 {
txn_id} 状态。") # 打印信息
return self._payment_gateway.get_transaction_status(txn_id) # 调用支付网关获取事务状态
# 客户端代码
print("
--- 遗留系统渐进式演进：适配器模式 ---")
legacy_gateway_instance = LegacyPaymentGateway() # 创建遗留支付网关实例
adapter = LegacyGatewayAdapter(legacy_gateway_instance) # 创建适配器，注入遗留网关
new_service = NewPaymentService(adapter) # 创建新服务，注入适配器
# 正常支付
print("
=== 正常支付 (通过适配器调用遗留网关) ===")
result = new_service.charge_customer(150.0, "1234567890123456", "12/25") # 扣款
print(f"支付结果: {
result}") # 打印结果
# 无效卡号 (由遗留网关拒绝)
print("
=== 无效卡号支付 (由遗留网关拒绝) ===")
result_invalid = new_service.charge_customer(50.0, "123", "12/25") # 扣款
print(f"支付结果: {
result_invalid}") # 打印结果
# 查询状态
print("
=== 查询支付状态 ===")
txn_id = result.get("new_txn_id") # 获取事务 ID
if txn_id: # 如果有事务 ID
status = new_service.check_payment_status(txn_id) # 检查支付状态
print(f"事务 {
txn_id} 的状态是: {
status}") # 打印状态
通过引入 LegacyGatewayAdapter，我们实现了：
DIP： NewPaymentService（高层模块）现在依赖于 NewPaymentGateway 抽象接口，而不是直接依赖 LegacyPaymentGateway 的具体实现。
OCP： 如果将来我们引入一个新的、完全现代化的支付网关（例如，StripeGateway），我们只需实现 NewPaymentGateway 接口，并将其注入到 NewPaymentService 中，而无需修改 NewPaymentService 本身。
SRP： LegacyGatewayAdapter 的单一职责是将 NewPaymentGateway 接口适配到 LegacyPaymentGateway。NewPaymentService 专注于业务扣款逻辑。
这种适配器模式是处理遗留系统技术债的强大策略。它允许你在不破坏现有代码的基础上，逐步引入遵循SOLID原则的新设计，从而实现架构的平滑演进。
8.5 SOLID原则在团队协作与开发流程中的实践
SOLID原则不仅是技术层面的指导，它们在团队协作和软件开发流程中也扮演着至关重要的角色。一个团队对SOLID原则的共同理解和一致实践，能够显著提升开发效率、代码质量和团队士气。
8.5.1 代码审查 (Code Review) 的强制力与学习机会
统一设计语言： 在团队中推广SOLID原则，使其成为一种共同的设计“语言”。在代码审查中，开发者可以基于这些原则进行讨论和反馈，而不是仅仅停留在表面问题。
早期发现问题： 代码审查是发现违反SOLID原则的有效阶段。例如，当一个类看起来过于庞大（可能违反SRP），或者依赖关系混乱（可能违反DIP）时，审查者可以及时指出，并在问题变得复杂之前进行修复。
知识共享与学习： 代码审查也是一个绝佳的学习机会。经验丰富的开发者可以通过审查来教育和引导新成员理解SOLID原则的应用。通过具体案例的讨论，将抽象的原则具象化。
审查检查点： 如前所述，为代码审查建立SOLID相关的检查点清单，可以系统化地确保原则的遵守。
示例：代码审查中的SOLID反馈
想象一个代码审查场景：
# 待审查的代码片段
class ProductManager: # 产品管理器
def __init__(self, db_client): # 构造函数
self.db = db_client # 数据库客户端
def create_product_and_notify_suppliers(self, product_data: dict, supplier_emails: list): # 创建产品并通知供应商
# 1. 验证产品数据
if not product_data.get("name") or product_data.get("price") <= 0: # 如果产品名称为空或价格小于等于0
raise ValueError("无效产品数据。") # 抛出值错误
# 2. 保存产品到数据库
product_id = self.db.insert_product(product_data) # 插入产品
print(f"产品 {
product_id} 已保存。") # 打印信息
# 3. 通知供应商
for email in supplier_emails: # 遍历供应商邮箱
self.db.send_email(email, f"新产品 {
product_data['name']} 已发布。", "请查看。") # 发送邮件
print("供应商已通知。") # 打印信息
return product_id # 返回产品ID
审查反馈 (基于SOLID)：
SRP 违反： ProductManager.create_product_and_notify_suppliers 方法承担了至少三个职责：产品数据验证、产品持久化、供应商通知。这使得这个方法过于复杂，未来任何一个职责的变化都可能导致它被修改。
建议：
将产品数据验证逻辑提取到 ProductValidator 类中。
将产品持久化操作（insert_product）封装到 ProductRepository 类中。
将供应商通知逻辑提取到 SupplierNotificationService 类中，该服务应依赖一个抽象的 EmailSender。
DIP 违反： ProductManager 直接依赖于具体的 db_client，并且 db_client 似乎同时承担了数据库操作和邮件发送的职责。这导致了高度耦合。
建议：
ProductManager（高层模块）不应直接依赖 db_client 的具体实现。它应该依赖一个 IProductRepository 接口和一个 ISupplierNotificationService 接口。
db_client 应该被拆分或适配成这些抽象接口的具体实现。
OCP 违反： 如果未来需要添加其他通知方式（如短信通知），create_product_and_notify_suppliers 方法将不得不修改其内部的通知逻辑。
建议： 引入一个 INotificationService 接口，并使用策略模式或工厂模式来支持多种通知方式。
通过这种方式的反馈，不仅指出了问题，还给出了基于SOLID原则的解决方案，帮助开发者提升设计能力。
8.5.2 结对编程 (Pair Programming) 与持续学习
实时讨论与设计决策： 结对编程提供了实时讨论设计决策的机会。当一个开发者开始编写一个可能违反SOLID的代码时，另一个开发者可以立即指出并共同探讨更好的解决方案。
最佳实践的传播： 经验丰富的开发者可以通过结对编程，将他们对SOLID原则的理解和实践经验，直接传授给经验较少的成员。
共同责任感： 结对编程使得代码质量成为团队的共同责任，而非个人。
8.5.3 持续集成/持续交付 (CI/CD) 中的质量门禁
自动化静态分析： 将前述的代码度量工具和Linter集成到CI/CD流程中作为质量门禁。例如，当一个提交引入的圈复杂度过高，或者类/函数行数超过阈值时，CI管道可以失败，强制开发者在合并代码前进行重构。
测试覆盖率： 自动化测试覆盖率的检查，鼓励开发者编写高质量的单元测试，这反过来又会促使他们编写更易于测试（即更遵循SOLID）的代码。
自动化重构建议： 某些更高级的CI工具甚至可以提供自动化重构建议，或与IDE集成，在开发阶段就提示潜在的SOLID违反。
8.5.4 领域特定语言 (DSL) 与SOLID
虽然不直接是SOLID的原则，但领域特定语言（DSL）可以帮助你更好地实现这些原则。
提高领域模型的清晰度： 通过DSL，你可以用更接近业务语言的方式表达业务规则和行为。这使得领域模型更加纯粹，更易于理解，从而间接促进了SRP。
隔离技术细节： DSL可以将底层的技术实现细节隐藏起来，让开发者专注于业务逻辑。这与DIP的目标一致，使得高层业务逻辑不必关心底层实现。
第九章：SOLID原则在Python生态系统中的应用：框架、库与最佳实践
Python拥有庞大而活跃的生态系统，包含众多优秀的框架和库。理解SOLID原则如何融入这些工具的设计中，以及如何在日常使用中利用它们来强化自己的SOLID实践，是成为Python面向对象设计专家的关键一步。本章将探讨常见Python框架（如Django、Flask）和库（如SQLAlchemy、Requests）如何体现SOLID，并提供在这些工具上构建遵循SOLID原则的应用的最佳实践。
9.1 Python Web框架中的SOLID原则
Python的Web框架在设计时普遍考虑了模块化、可扩展性和已关注点分离，这与SOLID原则高度契合。
9.1.1 Django与SOLID：MTV架构的体现
Django遵循MTV（Model-Template-View）架构模式，这与传统的MVC（Model-View-Controller）模式类似，但视图（View）在Django中扮演了控制器的角色，而模板（Template）则负责表示。
SRP (单一职责原则)：
Model： 负责数据结构、数据库交互和数据相关的业务逻辑。一个模型只应代表一个业务实体。
View： 接收HTTP请求，调用模型和表单（如果需要），处理业务逻辑，然后渲染模板或返回HTTP响应。它的职责是处理特定的URL请求。
Template： 负责数据展示，不应包含任何业务逻辑。
Forms： 负责数据验证和HTML表单渲染。
URL Dispatcher： 负责将URL请求路由到正确的视图。
这种分层和组件划分，清晰地体现了SRP，每个组件各司其职。
OCP (开放/封闭原则)：
Django的**可插拔应用（Pluggable Apps）**机制是OCP的绝佳体现。你可以开发独立的Django应用（如一个用户管理应用、一个博客应用），并在不同的项目中重用它们，而无需修改应用本身的源代码。
中间件 (Middleware)： 允许你在请求/响应生命周期中插入全局行为（如认证、会话管理），而无需修改视图或模型。这正是OCP的体现。
模板标签和过滤器： 允许你扩展模板的功能，而无需修改Django的核心模板引擎。
DIP (依赖倒置原则)：
虽然Django没有像DI容器那样显式地推广DIP，但它的组件设计鼓励DIP。例如，视图通过调用模型的方法与数据库交互，而不是直接执行SQL。模型内部可以抽象其数据库连接细节（例如，通过Django ORM）。
你可以在Django项目中定义抽象的业务逻辑接口，并在服务层（Service Layer）中注入依赖，使得业务逻辑不直接依赖模型或视图的具体实现。
LSP (里氏替换原则)：
Django的模型继承，尤其是使用抽象基类或多表继承时，如果设计得当，应遵循LSP。例如，一个通用的 Product 模型可以被 PhysicalProduct 或 DigitalProduct 安全地替换。
ISP (接口隔离原则)：
Django的不同组件通常提供特定用途的API。例如，模型管理器（Manager）只提供模型查询相关的方法，表单只提供验证和渲染相关的方法。
Django中的SOLID实践示例：用户注册
结合之前章节的SOLID实践，在Django中，我们可以这样组织用户注册功能：
# Django project structure:
# myproject/
# ├── myproject/
# │ └── settings.py
# └── users/ # Django App: 专注于用户管理
# ├── models.py # 模型
# ├── forms.py # 表单 (用于验证输入)
# ├── views.py # 视图 (扮演控制器角色)
# ├── services.py # 业务服务层 (自定义，非Django内置，推荐在此实现核心业务逻辑)
# ├── repositories.py # 数据仓库层 (自定义，推荐在此封装ORM操作)
# └── urls.py # URL 配置
# users/models.py (SRP for data definition)
from django.db import models # 导入 models 模块
class UserProfile(models.Model): # 用户档案模型
username = models.CharField(max_length=150, unique=True) # 用户名
email = models.EmailField(unique=True) # 邮箱
password_hash = models.CharField(max_length=128) # 密码哈希
is_active = models.BooleanField(default=False) # 是否激活
created_at = models.DateTimeField(auto_now_add=True) # 创建时间
def __str__(self): # 字符串表示
return self.username # 返回用户名
# users/forms.py (SRP for input validation and form rendering)
from django import forms # 导入 forms 模块
class UserRegistrationForm(forms.Form): # 用户注册表单
username = forms.CharField(max_length=150) # 用户名
email = forms.EmailField() # 邮箱
password = forms.CharField(widget=forms.PasswordInput) # 密码
def clean_username(self): # 清理用户名
username = self.cleaned_data['username'] # 获取用户名
# 实际这里会通过 UserRepository 检查唯一性，这里简化
if username == "admin": # 如果用户名是 admin
raise forms.ValidationError("用户名'admin'已被占用。") # 抛出验证错误
return username # 返回用户名
# users/repositories.py (SRP for ORM operations, DIP via abstracting ORM)
from typing import Protocol, Optional # 导入 Protocol, Optional
# 定义抽象仓库接口 (DIP)
class IUserRepository(Protocol): # 用户数据仓库协议
def get_by_username(self, username: str) -> Optional[UserProfile]: # 根据用户名获取用户
...
def get_by_email(self, email: str) -> Optional[UserProfile]: # 根据邮箱获取用户
...
def create_user(self, username: str, email: str, password_hash: str) -> UserProfile: # 创建用户
...
def save_user(self, user: UserProfile): # 保存用户
...
# 具体的Django ORM实现 (DIP的实现者)
class DjangoUserRepository: # Django 用户数据仓库
def get_by_username(self, username: str) -> Optional[UserProfile]: # 实现根据用户名获取用户
print(f"Repo: 从Django ORM获取用户 '{
username}'。") # 打印信息
return UserProfile.objects.filter(username=username).first() # 返回第一个匹配的用户
def get_by_email(self, email: str) -> Optional[UserProfile]: # 实现根据邮箱获取用户
print(f"Repo: 从Django ORM获取邮箱 '{
email}'。") # 打印信息
return UserProfile.objects.filter(email=email).first() # 返回第一个匹配的用户
def create_user(self, username: str, email: str, password_hash: str) -> UserProfile: # 实现创建用户
print(f"Repo: 通过Django ORM创建用户 '{
username}'。") # 打印信息
# 实际这里应该使用Django内置的User模型和密码哈希函数
user = UserProfile(username=username, email=email, password_hash=password_hash) # 创建用户
user.save() # 保存用户
return user # 返回用户
def save_user(self, user: UserProfile): # 实现保存用户
print(f"Repo: 通过Django ORM保存用户 '{
user.username}'。") # 打印信息
user.save() # 保存用户
# users/services.py (SRP for business logic, DIP for dependencies)
from .repositories import IUserRepository, DjangoUserRepository # 从当前包导入 IUserRepository, DjangoUserRepository
# 假设邮件服务已定义并遵循DIP
class IEmailService(Protocol): # 邮件服务协议
def send_welcome_email(self, recipient_email: str, username: str): # 发送欢迎邮件
...
class MockEmailService: # 模拟邮件服务
def send_welcome_email(self, recipient_email: str, username: str): # 实现发送欢迎邮件
print(f"MockEmailService: 模拟发送欢迎邮件到 {
recipient_email}。") # 打印信息
class UserRegistrationService: # 用户注册服务
def __init__(self, user_repo: IUserRepository, email_service: IEmailService): # 构造函数
self._user_repo = user_repo # 注入用户数据仓库
self._email_service = email_service # 注入邮件服务
def register_new_user(self, username: str, email: str, password_raw: str) -> UserProfile: # 注册新用户
# 业务规则：检查用户名和邮箱唯一性
if self._user_repo.get_by_username(username): # 如果用户名已存在
raise ValueError("用户名已存在。") # 抛出值错误
if self._user_repo.get_by_email(email): # 如果邮箱已存在
raise ValueError("邮箱已存在。") # 抛出值错误
# 密码哈希 (实际用Django的make_password)
password_hash = f"hashed_{
password_raw}" # 密码哈希
new_user = self._user_repo.create_user(username, email, password_hash) # 创建用户
self._email_service.send_welcome_email(email, username) # 发送欢迎邮件
print(f"Service: 用户 {
username} 注册成功。") # 打印信息
return new_user # 返回新用户
# users/views.py (SRP for HTTP request/response, DIP for business service)
from django.shortcuts import render, redirect # 导入 render, redirect
from django.http import HttpRequest, HttpResponse, JsonResponse # 导入 HttpRequest, HttpResponse, JsonResponse
from django.views import View # 导入 View
from .forms import UserRegistrationForm # 导入 UserRegistrationForm
from .services import UserRegistrationService, DjangoUserRepository, MockEmailService # 导入 UserRegistrationService, DjangoUserRepository, MockEmailService
# 使用全局变量或DI容器来管理服务实例 (Django通常会集成DI容器)
# 这里简化为直接实例化
user_repo_instance = DjangoUserRepository() # 创建 Django 用户数据仓库实例
email_service_instance = MockEmailService() # 创建模拟邮件服务实例
user_registration_service = UserRegistrationService(user_repo_instance, email_service_instance) # 创建用户注册服务实例
class RegisterUserView(View): # 注册用户视图类
def get(self, request: HttpRequest) -> HttpResponse: # GET 请求
form = UserRegistrationForm() # 创建表单
return render(request, 'users/register.html', {
'form': form}) # 渲染注册页面
def post(self, request: HttpRequest) -> HttpResponse: # POST 请求
form = UserRegistrationForm(request.POST) # 使用 POST 数据创建表单
if form.is_valid(): # 如果表单有效
try: # 尝试注册用户
username = form.cleaned_data['username'] # 获取用户名
email = form.cleaned_data['email'] # 获取邮箱
password = form.cleaned_data['password'] # 获取密码
user = user_registration_service.register_new_user(username, email, password) # 调用用户注册服务
print(f"View: 用户 '{
user.username}' 已通过视图注册。") # 打印信息
# return redirect('success_page') # 重定向到成功页面
return JsonResponse({
"status": "success", "user_id": user.pk, "message": "注册成功！"}, status=201) # 返回 JSON 响应
except ValueError as e: # 捕获值错误
form.add_error(None, str(e)) # 添加错误到表单
# 如果表单无效或有业务错误
return JsonResponse({
"status": "error", "errors": form.errors}, status=400) # 返回 JSON 响应
# users/urls.py
from django.urls import path # 导入 path
from .views import RegisterUserView # 导入 RegisterUserView
urlpatterns = [ # URL 模式列表
path('register/', RegisterUserView.as_view(), name='register'), # 注册 URL
]
这个Django示例展示了：
SRP： 模型只定义数据，表单只处理验证和渲染，仓库只处理ORM操作，服务只处理业务逻辑，视图只处理HTTP请求和响应。
DIP： UserRegistrationService 依赖于 IUserRepository 和 IEmailService 协议，而不是 DjangoUserRepository 或 MockEmailService 的具体实现。这使得服务层与具体的ORM实现和邮件发送机制解耦。
OCP： 如果将来需要更换ORM（例如，从Django ORM切换到SQLAlchemy）或邮件服务提供商，只需创建新的 IUserRepository 或 IEmailService 实现，并修改依赖注入配置，而 UserRegistrationService 和 RegisterUserView 保持不变。
9.1.2 Flask与SOLID：微框架的灵活性
Flask作为一个微框架，提供了更大的自由度来组织项目结构。这种灵活性意味着开发者有更多机会（也承担更多责任）来主动应用SOLID原则。
SRP： Flask不强制任何特定的组件划分，但其轻量级特性鼓励你将不同的职责（如请求处理、业务逻辑、数据访问）分离到不同的模块或蓝图（Blueprints）中。你可以手动创建服务层、仓库层等。
OCP： Flask的蓝图机制是OCP的体现。你可以创建独立的、可重用的蓝图来封装特定功能，并将其“插拔”到不同的Flask应用中，而无需修改蓝图本身的源代码。这允许你通过添加蓝图来扩展应用功能。
DIP： Flask没有内置的DI容器，但你可以集成第三方DI库（如Flask-Injector, Flask-Dependency-Injector），或者手动通过构造函数注入来实践DIP。这使得你的业务逻辑可以依赖抽象，而不是具体的Web框架组件。
LSP： 如果你使用Flask扩展（Extensions）或自定义组件，这些组件的实现应该能够替换其抽象，而不会破坏框架的预期行为。
ISP： 在Flask中，你可以根据需要定义和使用小而精的函数或类来处理特定任务，避免创建大而全的接口或组件。
Flask中的SOLID实践示例：任务管理API
# Flask project structure (example with service/repository pattern):
# my_flask_app/
# ├── app.py # Flask 应用入口
# ├── config.py # 配置
# ├── core/
# │ ├── models.py # 领域模型
# │ └── interfaces.py # 抽象接口 (DIP)
# ├── repositories/
# │ ├── __init__.py
# │ └── in_memory_repo.py # 内存实现
# │ └── sqlalchemy_repo.py # SQLAlchemy 实现
# ├── services/
# │ ├── __init__.py
# │ └── task_service.py # 业务服务
# └── api/
# ├── __init__.py
# └── task_api.py # API 接口 (Flask Blueprint)
# core/models.py
import uuid # 导入 uuid 模块
class Task: # 任务类
def __init__(self, title: str, description: str, status: str = "PENDING", task_id: str = None): # 构造函数
self.task_id = task_id if task_id else str(uuid.uuid4()) # 任务ID
self.title = title # 标题
self.description = description # 描述
self.status = status # 状态
def __repr__(self): # 开发者友好的字符串表示
return f"Task(id='{
self.task_id}', title='{
self.title}', status='{
self.status}')" # 返回 Task 对象的可重建字符串
def to_dict(self): # 转换为字典
return {
"id": self.task_id, "title": self.title, "description": self.description, "status": self.status} # 返回字典
# core/interfaces.py (DIP)
import abc # 导入 abc 模块
from typing import List, Optional, Protocol # 导入 List, Optional, Protocol
class ITaskRepository(Protocol): # 任务数据仓库协议
def get_all(self) -> List[Task]: # 获取所有任务
...
def get_by_id(self, task_id: str) -> Optional[Task]: # 根据 ID 获取任务
...
def add(self, task: Task): # 添加任务
...
def update(self, task: Task): # 更新任务
...
def delete(self, task_id: str): # 删除任务
...
# repositories/in_memory_repo.py (SRP: In-memory persistence)
from core.interfaces import ITaskRepository # 导入 ITaskRepository
from core.models import Task # 导入 Task
from typing import Dict # 导入 Dict
class InMemoryTaskRepository: # 内存任务数据仓库
def __init__(self): # 构造函数
self._tasks: Dict[str, Task] = {
} # 任务字典
def get_all(self) -> List[Task]: # 实现获取所有任务
print("InMemoryRepo: 获取所有任务。") # 打印信息
return list(self._tasks.values()) # 返回所有任务的列表
def get_by_id(self, task_id: str) -> Optional[Task]: # 实现根据 ID 获取任务
print(f"InMemoryRepo: 获取任务 {
task_id}。") # 打印信息
return self._tasks.get(task_id) # 返回任务
def add(self, task: Task): # 实现添加任务
print(f"InMemoryRepo: 添加任务 {
task.task_id}。") # 打印信息
if task.task_id in self._tasks: # 如果任务已存在
raise ValueError("任务已存在。") # 抛出值错误
self._tasks[task.task_id] = task # 添加任务
def update(self, task: Task): # 实现更新任务
print(f"InMemoryRepo: 更新任务 {
task.task_id}。") # 打印信息
if task.task_id not in self._tasks: # 如果任务不存在
raise ValueError("任务不存在，无法更新。") # 抛出值错误
self._tasks
暂无评论内容