Python设计模式应用最佳实践总结

嘿，Python学习搭子！在掌握了设计模式的基本概念和实现之后，咱们今天要来聊聊最重要的一课——如何在实际项目中正确应用设计模式。就像学会了烹饪技巧，还要知道什么时候用大火、什么时候用小火一样，设计模式的应用也需要遵循一些最佳实践。

1. 1. 明智选择模式：根据场景选择最合适的模式，避免"为了模式而模式"
2. 2. Pythonic实现：用Python特有的方式优雅实现设计模式
3. 3. 规避常见陷阱：识别并避免设计模式应用中的典型错误
4. 4. 编写可测试代码：让设计模式友好的代码更容易测试
5. 5. 团队协作规范：建立团队内的设计模式使用约定

# 反例：简单需求过度使用设计模式
classUser:
def__init__(self, name, email):
self.name = name
self.email = email

# 不必要的工厂模式
classUserFactory:
    @staticmethod
defcreate(name, email):
return User(name, email)

# 不必要的仓储模式
classUserRepository:
def__init__(self):
self.users = {}

defsave(self, user):
self.users[user.name] = user

# 实际使用
user = UserFactory.create("张三", "zhangsan@example.com")
repo = UserRepository()
repo.save(user)

# 正例：直接使用简单结构
user = {"name": "张三", "email": "zhangsan@example.com"}
users = {}  # 简单的字典存储
users[user["name"]] = user

# 或者使用数据类
from dataclasses import dataclass

@dataclass
classUser:
    name: str
    email: str

user = User("张三", "zhangsan@example.com")

1. 问题是否复杂？
   ├── 否 → 直接实现，保持简单
   └── 是 → 继续

2. 是否有明确的设计模式适用？
   ├── 否 → 考虑其他架构方案
   └── 是 → 继续

3. 引入模式的收益 > 复杂度成本？
   ├── 否 → 简化方案
   └── 是 → 使用该模式

# 组合示例：工厂 + 策略 + 装饰器
classPaymentMethod:
"""支付策略接口"""
defpay(self, amount: float) -> str:
pass

classAlipay(PaymentMethod):
defpay(self, amount: float) -> str:
returnf"支付宝支付 {amount}"

classWechatPay(PaymentMethod):
defpay(self, amount: float) -> str:
returnf"微信支付 {amount}"

# 装饰器：添加日志
deflog_payment(func):
defwrapper(self, amount):
print(f"[LOG] 开始支付: {amount}")
        result = func(self, amount)
print(f"[LOG] 支付完成: {result}")
return result
return wrapper

classLoggedWechatPay(WechatPay):
    @log_payment
defpay(self, amount: float) -> str:
returnsuper().pay(amount)

# 工厂：创建支付实例
classPaymentFactory:
    @staticmethod
defcreate(method: str) -> PaymentMethod:
if method == "alipay":
return Alipay()
elif method == "wechat":
return LoggedWechatPay()  # 返回带日志的版本
else:
raise ValueError(f"不支持的支付方式: {method}")

# 使用
payment = PaymentFactory.create("wechat")
result = payment.pay(100.0)

# 传统策略模式（面向对象）
classDiscountStrategy:
defcalculate(self, price: float) -> float:
pass

classNoDiscount(DiscountStrategy):
defcalculate(self, price: float) -> float:
return price

classPercentageDiscount(DiscountStrategy):
def__init__(self, percentage: float):
self.percentage = percentage

defcalculate(self, price: float) -> float:
return price * (1 - self.percentage / 100)

# Pythonic策略模式（使用函数）
defno_discount(price: float) -> float:
return price

defpercentage_discount(percentage: float):
defdiscount_func(price: float) -> float:
return price * (1 - percentage / 100)
return discount_func

# 使用字典映射策略
discount_strategies = {
"none": no_discount,
"10%": percentage_discount(10),
"20%": percentage_discount(20)
}

defcalculate_price(price: float, strategy_name: str) -> float:
    strategy = discount_strategies.get(strategy_name, no_discount)
return strategy(price)

# 测试
print(calculate_price(100.0, "10%"))  # 90.0
print(calculate_price(100.0, "none"))  # 100.0

# 传统装饰器模式
classComponent:
defoperation(self) -> str:
return"原始组件"

classDecorator(Component):
def__init__(self, component: Component):
self._component = component

defoperation(self) -> str:
returnself._component.operation()

classConcreteDecoratorA(Decorator):
defoperation(self) -> str:
returnf"装饰器A({super().operation()})"

# 使用
component = Component()
decorated = ConcreteDecoratorA(component)
print(decorated.operation())  # 装饰器A(原始组件)

# Pythonic装饰器模式
defcomponent() -> str:
return"原始组件"

defdecorator_a(func):
defwrapper():
returnf"装饰器A({func()})"
return wrapper

@decorator_a
defdecorated_component() -> str:
return component()

print(decorated_component())  # 装饰器A(原始组件)

# 传统建造者模式
classDatabaseConnectionBuilder:
def__init__(self):
self._config = {}

defset_host(self, host: str):
self._config["host"] = host
returnself

defset_port(self, port: int):
self._config["port"] = port
returnself

defset_timeout(self, timeout: float):
self._config["timeout"] = timeout
returnself

defbuild(self):
return DatabaseConnection(**self._config)

# 使用
conn = (DatabaseConnectionBuilder()
        .set_host("localhost")
        .set_port(5432)
        .set_timeout(30.0)
        .build())

# Pythonic方式：使用上下文管理器
import contextlib
from contextlib import contextmanager

@contextmanager
defdatabase_connection(host: str = "localhost", 
                       port: int = 5432, 
                       timeout: float = 30.0):
"""创建数据库连接的上下文管理器"""
    conn = DatabaseConnection(host=host, port=port, timeout=timeout)
try:
        conn.connect()
yield conn
finally:
        conn.disconnect()

# 使用
with database_connection(host="db.example.com", timeout=60.0) as conn:
    results = conn.execute_query("SELECT * FROM users")

from dataclasses import dataclass, field
from typing importList, Optional
from datetime import datetime

# 传统方式
classTraditionalOrder:
def__init__(self, order_id: str, customer_id: str, items: List[str], 
                 total: float, created_at: datetime = None):
self.order_id = order_id
self.customer_id = customer_id
self.items = items
self.total = total
self.created_at = created_at or datetime.now()

def__repr__(self):
returnf"Order({self.order_id}, {self.customer_id}, {self.total})"

def__eq__(self, other):
ifnotisinstance(other, TraditionalOrder):
returnFalse
returnself.order_id == other.order_id

# 数据类方式
@dataclass(order=True)  # 启用排序
classOrder:
    order_id: str
    customer_id: str
    items: List[str] = field(default_factory=list)
    total: float = 0.0
    created_at: datetime = field(default_factory=datetime.now)
    status: str = "created"

    @property
defitem_count(self) -> int:
returnlen(self.items)

defapply_discount(self, percentage: float):
"""应用折扣"""
        discount = self.total * percentage / 100
self.total -= discount
returnself

# 使用数据类的优势
order1 = Order("ORD001", "CUST001", ["Book"], 50.0)
order2 = Order("ORD002", "CUST002", ["Pen"], 10.0)

print(order1)  # 自动生成__repr__
print(order1 == order2)  # 自动生成__eq__
print(order1.item_count)  # 自定义属性

# 数据类还支持：
# - 不可变实例：@dataclass(frozen=True)
# - 后初始化：__post_init__方法
# - 字段元数据：field(metadata={"unit": "USD"})

from typing import Protocol, runtime_checkable
from dataclasses import dataclass

# 传统方式：抽象基类
from abc import ABC, abstractmethod

classNotificationService(ABC):
    @abstractmethod
defsend(self, message: str, recipient: str) -> bool:
pass

classEmailService(NotificationService):
defsend(self, message: str, recipient: str) -> bool:
print(f"发送邮件给 {recipient}: {message}")
returnTrue

# 协议方式：更灵活
@runtime_checkable
classNotificationProtocol(Protocol):
defsend(self, message: str, recipient: str) -> bool:
        ...

# 任何实现了send方法的类都符合协议
classSMSService:
defsend(self, message: str, phone_number: str) -> bool:
print(f"发送短信给 {phone_number}: {message}")
returnTrue

classPushNotificationService:
defsend(self, content: str, device_token: str) -> bool:
print(f"推送通知到设备 {device_token}: {content}")
returnTrue

# 协议检查
defcheck_notification_service(service: NotificationProtocol):
returnisinstance(service, NotificationProtocol)

# 使用
sms = SMSService()
print(check_notification_service(sms))  # True，因为SMSService有send方法

push = PushNotificationService()
print(check_notification_service(push))  # True

# 鸭子类型的好处：不需要显式继承

# 传统迭代器模式
classFibonacciIterator:
def__init__(self, max_count: int = 10):
self.max_count = max_count
self.count = 0
self.a, self.b = 0, 1

def__iter__(self):
returnself

def__next__(self):
ifself.count >= self.max_count:
raise StopIteration

        result = self.a
self.a, self.b = self.b, self.a + self.b
self.count += 1
return result

# Pythonic方式：生成器
deffibonacci_generator(max_count: int = 10):
    a, b = 0, 1
    count = 0
while count < max_count:
yield a
        a, b = b, a + b
        count += 1

# 使用
for num in fibonacci_generator(5):
print(num)

# 协程实现状态机
asyncdeforder_state_machine():
"""订单状态机的协程实现"""
    state = "created"

whileTrue:
        event = yield state

if state == "created"and event == "pay":
            state = "paid"
print("订单已支付")

elif state == "paid"and event == "ship":
            state = "shipped"
print("订单已发货")

elif state == "shipped"and event == "deliver":
            state = "delivered"
print("订单已送达")

else:
print(f"无效状态转换: {state} -> {event}")

# 使用
asyncdefmain():
    state_machine = order_state_machine()
await state_machine.asend(None)  # 初始化

for event in ["pay", "ship", "deliver", "invalid"]:
        state = await state_machine.asend(event)
print(f"当前状态: {state}")

# asyncio.run(main())

# 错误：处处使用单例
classConfigSingleton:
    _instance = None

def__new__(cls):
if cls._instance isNone:
            cls._instance = super().__new__(cls)
return cls._instance

def__init__(self):
self.settings = {}

classLoggerSingleton:
    _instance = None

def__new__(cls):
if cls._instance isNone:
            cls._instance = super().__new__(cls)
return cls._instance

classCacheSingleton:
    _instance = None

def__new__(cls):
if cls._instance isNone:
            cls._instance = super().__new__(cls)
return cls._instance

# 正确：根据需求选择
# 1. 配置：可以使用模块级变量
config = {"debug": True, "timeout": 30}

# 2. 日志：使用标准库logging模块
import logging
logging.basicConfig(level=logging.INFO)

# 3. 缓存：根据需要选择
from functools import lru_cache

@lru_cache(maxsize=128)
defexpensive_computation(n):
return n * n

# 错误：过度复杂的抽象层次
classAbstractNotification:
defsend(self, message: str):
pass

classEmailNotification(AbstractNotification):
pass

classDecoratedEmailNotification(EmailNotification):
pass

classNotificationFactory:
defcreate(self, type: str) -> AbstractNotification:
pass

classNotificationManager:
def__init__(self, factory: NotificationFactory):
self.factory = factory

defnotify(self, type: str, message: str):
        notification = self.factory.create(type)
        notification.send(message)

# 正确：简化设计
defsend_email(message: str, recipient: str):
print(f"发送邮件给 {recipient}: {message}")

defsend_sms(message: str, phone: str):
print(f"发送短信给 {phone}: {message}")

notification_functions = {
"email": send_email,
"sms": send_sms
}

defnotify(type: str, message: str, recipient: str):
    func = notification_functions.get(type)
if func:
        func(message, recipient)

import time
from functools import wraps

# 多个装饰器嵌套
deftiming_decorator(func):
    @wraps(func)
defwrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
print(f"{func.__name__} 耗时: {time.time() - start:.4f}秒")
return result
return wrapper

deflogging_decorator(func):
    @wraps(func)
defwrapper(*args, **kwargs):
print(f"调用 {func.__name__}")
return func(*args, **kwargs)
return wrapper

defvalidation_decorator(func):
    @wraps(func)
defwrapper(*args, **kwargs):
ifnot args:
raise ValueError("参数不能为空")
return func(*args, **kwargs)
return wrapper

# 三层装饰器
@timing_decorator
@logging_decorator
@validation_decorator
defcomplex_calculation(n):
returnsum(range(n))

# 每次调用都有三层函数调用开销
# 解决方案：合并装饰器或使用类装饰器

# 错误：频繁创建策略对象
defprocess_data(data, strategy_name):
if strategy_name == "strategy_a":
        strategy = StrategyA()
elif strategy_name == "strategy_b":
        strategy = StrategyB()
else:
        strategy = DefaultStrategy()

return strategy.process(data)

# 频繁调用时，每次都要创建新对象
for i inrange(10000):
    process_data(data, "strategy_a")

# 正确：复用对象
classStrategyFactory:
def__init__(self):
self._strategies = {
"strategy_a": StrategyA(),
"strategy_b": StrategyB(),
"default": DefaultStrategy()
        }

defget_strategy(self, name: str):
returnself._strategies.get(name, self._strategies["default"])

factory = StrategyFactory()

defoptimized_process_data(data, strategy_name):
    strategy = factory.get_strategy(strategy_name)
return strategy.process(data)

# 错误：没有文档的复杂模式
classA:
def__init__(self, b):
self.b = b
self.c = C()
self.d = D(self.b, self.c)
self.e = E(self.d)

# 正确：添加文档和类型提示
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from .b_module import B
from .c_module import C
from .d_module import D
from .e_module import E

classServiceCoordinator:
"""
    服务协调器：组合多个服务完成复杂业务

    设计模式：
    1. 外观模式：对外提供统一接口
    2. 依赖注入：通过构造函数注入依赖
    3. 组合模式：组合多个服务

    依赖关系：
    ServiceCoordinator → B → C
                     ↓
                     D → E
    """

def__init__(self, b_service: 'B', c_service: 'C'):
self.b = b_service
self.c = c_service
self.d = D(b_service, c_service)
self.e = E(self.d)

# 错误：观察者模式导致的隐式耦合
classOrderService:
def__init__(self):
self._observers = []  # 内部维护观察者

defcreate_order(self, data):
        order = self._save_order(data)

# 隐式调用多个服务
for observer inself._observers:
            observer.on_order_created(order)

return order

# 正确：显式依赖，更易理解
classOrderService:
def__init__(self, 
                 inventory_service: InventoryService,
                 payment_service: PaymentService,
                 notification_service: NotificationService):
self.inventory = inventory_service
self.payment = payment_service
self.notification = notification_service

defcreate_order(self, data):
        order = self._save_order(data)

# 显式调用
self.inventory.reserve_stock(order)
self.payment.process_payment(order)
self.notification.send_confirmation(order)

return order

from typing import Protocol
import unittest.mock as mock

# 定义协议
classPaymentGateway(Protocol):
defcharge(self, amount: float, token: str) -> dict:
        ...

# 业务类使用依赖注入
classOrderProcessor:
def__init__(self, payment_gateway: PaymentGateway):
self.payment = payment_gateway

defprocess_order(self, order_data: dict) -> bool:
# 使用注入的依赖
        result = self.payment.charge(
            order_data["amount"],
            order_data["payment_token"]
        )
return result.get("success", False)

# 测试：使用Mock对象
deftest_order_processor():
# 创建Mock支付网关
    mock_gateway = mock.Mock(spec=PaymentGateway)
    mock_gateway.charge.return_value = {"success": True}

# 注入Mock
    processor = OrderProcessor(mock_gateway)

# 测试数据
    order_data = {
"amount": 100.0,
"payment_token": "tok_123"
    }

# 执行测试
    result = processor.process_order(order_data)

# 验证
assert result isTrue
    mock_gateway.charge.assert_called_once_with(100.0, "tok_123")

from abc import ABC, abstractmethod
import pytest

# 抽象产品
classFormatter(ABC):
    @abstractmethod
defformat(self, data: str) -> str:
pass

# 具体产品
classJSONFormatter(Formatter):
defformat(self, data: str) -> str:
import json
return json.dumps({"data": data})

classXMLFormatter(Formatter):
defformat(self, data: str) -> str:
returnf"<data>{data}</data>"

# 工厂
classFormatterFactory:
    @staticmethod
defcreate(format_type: str) -> Formatter:
if format_type == "json":
return JSONFormatter()
elif format_type == "xml":
return XMLFormatter()
else:
raise ValueError(f"不支持的格式: {format_type}")

# 测试工厂
deftest_formatter_factory():
# 测试JSON格式化器
    json_formatter = FormatterFactory.create("json")
    result = json_formatter.format("test")
assert result == '{"data": "test"}'

# 测试XML格式化器
    xml_formatter = FormatterFactory.create("xml")
    result = xml_formatter.format("test")
assert result == "<data>test</data>"

# 测试无效类型
with pytest.raises(ValueError):
        FormatterFactory.create("invalid")

# 使用pytest的参数化测试
@pytest.mark.parametrize("format_type, input_data, expected", [
    ("json", "hello", '{"data": "hello"}'),
    ("xml", "world", "<data>world</data>"),
])
deftest_formatters(format_type, input_data, expected):
    formatter = FormatterFactory.create(format_type)
    result = formatter.format(input_data)
assert result == expected

import asyncio
from typing importSet, Callable, Any
import pytest

classObservable:
"""可测试的观察者模式实现"""

def__init__(self):
self._observers: Set[Callable] = set()

defsubscribe(self, observer: Callable):
self._observers.add(observer)

defunsubscribe(self, observer: Callable):
self._observers.discard(observer)

defnotify(self, event_data: Any):
for observer inself._observers:
            observer(event_data)

    @property
defobserver_count(self) -> int:
returnlen(self._observers)

# 测试用例
classTestObservable:
defsetup_method(self):
self.observable = Observable()

deftest_subscribe(self):
defobserver(data):
pass

self.observable.subscribe(observer)
assertself.observable.observer_count == 1

deftest_unsubscribe(self):
defobserver(data):
pass

self.observable.subscribe(observer)
self.observable.unsubscribe(observer)
assertself.observable.observer_count == 0

deftest_notify(self):
        received_data = []

defobserver(data):
            received_data.append(data)

self.observable.subscribe(observer)
self.observable.notify("test_event")

assert received_data == ["test_event"]

deftest_multiple_observers(self):
        results = []

defobserver1(data):
            results.append(f"observer1: {data}")

defobserver2(data):
            results.append(f"observer2: {data}")

self.observable.subscribe(observer1)
self.observable.subscribe(observer2)
self.observable.notify("event")

assertlen(results) == 2
assert"observer1: event"in results
assert"observer2: event"in results

from typing importList, Callable
import pytest

# 策略接口
SortingStrategy = Callable[[List[int]], List[int]]

# 具体策略
defbubble_sort(numbers: List[int]) -> List[int]:
"""冒泡排序"""
    n = len(numbers)
for i inrange(n):
for j inrange(0, n - i - 1):
if numbers[j] > numbers[j + 1]:
                numbers[j], numbers[j + 1] = numbers[j + 1], numbers[j]
return numbers

defquick_sort(numbers: List[int]) -> List[int]:
"""快速排序"""
iflen(numbers) <= 1:
return numbers

    pivot = numbers[len(numbers) // 2]
    left = [x for x in numbers if x < pivot]
    middle = [x for x in numbers if x == pivot]
    right = [x for x in numbers if x > pivot]

return quick_sort(left) + middle + quick_sort(right)

# 上下文
classSorter:
def__init__(self, strategy: SortingStrategy):
self.strategy = strategy

defsort(self, numbers: List[int]) -> List[int]:
returnself.strategy(numbers.copy())  # 避免修改原数据

defset_strategy(self, strategy: SortingStrategy):
self.strategy = strategy

# 测试
classTestSorter:
    @pytest.fixture
defunsorted_list(self):
return [64, 34, 25, 12, 22, 11, 90]

    @pytest.fixture
defsorted_list(self):
return [11, 12, 22, 25, 34, 64, 90]

deftest_bubble_sort(self, unsorted_list, sorted_list):
        sorter = Sorter(bubble_sort)
        result = sorter.sort(unsorted_list)
assert result == sorted_list

deftest_quick_sort(self, unsorted_list, sorted_list):
        sorter = Sorter(quick_sort)
        result = sorter.sort(unsorted_list)
assert result == sorted_list

deftest_strategy_switching(self, unsorted_list, sorted_list):
        sorter = Sorter(bubble_sort)
        result1 = sorter.sort(unsorted_list)

        sorter.set_strategy(quick_sort)
        result2 = sorter.sort(unsorted_list)

assert result1 == sorted_list
assert result2 == sorted_list

# .coveragerc 配置文件
[run]
source = .
omit =
    */tests/*
    */migrations/*
    */__pycache__/*
    */venv/*
    */.venv/*

[report]
exclude_lines =
    pragma: no cover
def__repr__
ifself.debug
if settings.DEBUG
raise AssertionError
raise NotImplementedError
if0:
if __name__ == .__main__.:
class .*\bProtocol\b:
    @.*abc.abstractmethod

# pytest配置文件：pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"--verbose",
"--tb=short",
"--strict-markers",
"--cov=.",
"--cov-report=term-missing",
"--cov-report=html"
]
markers = [
"unit: 单元测试",
"integration: 集成测试",
"slow: 慢速测试",
"design_pattern: 设计模式相关测试"
]

# tests/test_design_patterns.py
import pytest
from design_patterns.factory import PaymentFactory
from design_patterns.strategy import Sorter, bubble_sort, quick_sort
from design_patterns.observer import Observable

classTestDesignPatterns:
"""设计模式测试套件"""

    @pytest.mark.design_pattern
    @pytest.mark.factory
deftest_payment_factory(self):
"""测试工厂模式"""
        alipay = PaymentFactory.create("alipay")
assert alipay.process(100) isTrue

        wechat = PaymentFactory.create("wechat")
assert wechat.process(200) isTrue

with pytest.raises(ValueError):
            PaymentFactory.create("invalid")

    @pytest.mark.design_pattern
    @pytest.mark.strategy
deftest_sorting_strategies(self):
"""测试策略模式"""
        sorter = Sorter(bubble_sort)
        result = sorter.sort([3, 1, 4, 1, 5])
assert result == [1, 1, 3, 4, 5]

        sorter.set_strategy(quick_sort)
        result = sorter.sort([3, 1, 4, 1, 5])
assert result == [1, 1, 3, 4, 5]

    @pytest.mark.design_pattern
    @pytest.mark.observer
deftest_observer_pattern(self):
"""测试观察者模式"""
        observable = Observable()
        events = []

defobserver(data):
            events.append(data)

        observable.subscribe(observer)
        observable.notify("event1")

assert events == ["event1"]

        observable.unsubscribe(observer)
        observable.notify("event2")

assert events == ["event1"]  # 取消订阅后不应收到事件

# 运行特定标记的测试
# pytest -m "design_pattern and factory" -v
# pytest -m "design_pattern and strategy" -v

# 工厂模式
classPaymentFactory:          # 主工厂类
classPaymentCreator:          # 替代命名
classPaymentBuilder:          # 建造者模式

# 策略模式
classSortingStrategy:         # 策略接口
classQuickSortStrategy:       # 具体策略
classBubbleSortStrategy:      # 具体策略

# 观察者模式
classOrderSubject:            # 被观察者
classOrderObserver:           # 观察者接口
classEmailObserver:           # 具体观察者

# 装饰器模式
classLoggerDecorator:         # 装饰器类
classTimedOperation:          # 功能描述

# 适配器模式
classLegacySystemAdapter:     # 适配器类
classNewInterfaceAdapter:     # 适配器类

classNotificationService:
# 观察者模式方法
defsubscribe(self, observer):     # 订阅
defunsubscribe(self, observer):   # 取消订阅
defnotify(self, event):           # 通知

# 工厂模式方法
    @classmethod
defcreate(cls, type):             # 创建实例
    @classmethod
deffrom_config(cls, config):      # 从配置创建

# 策略模式方法
defset_strategy(self, strategy):  # 设置策略
defexecute(self, data):           # 执行策略

# 设计模式代码审查清单

## 1. 模式选择合理性
- [ ] 是否真的需要设计模式？
- [ ] 所选模式是否最适合当前场景？
- [ ] 是否有更简单的解决方案？

## 2. 实现正确性
- [ ] 模式的核心概念是否正确实现？
- [ ] 是否符合Pythonic原则？
- [ ] 是否有不必要的复杂性？

## 3. 可测试性
- [ ] 代码是否易于单元测试？
- [ ] 依赖是否清晰可注入？
- [ ] Mock对象是否容易创建？

## 4. 可维护性
- [ ] 命名是否清晰表达意图？
- [ ] 文档是否充分说明设计模式的使用？
- [ ] 是否有过度抽象或模式嵌套？

## 5. 性能考量
- [ ] 模式引入的开销是否可以接受？
- [ ] 是否有不必要的对象创建？
- [ ] 是否有性能热点？

# 反模式识别代码示例
defdetect_anti_patterns(code_snippet: str) -> List[str]:
"""检测设计模式反模式"""
    anti_patterns = []

# 1. 过度使用单例模式
if re.search(r"class .*Singleton.*:", code_snippet) and \
       re.search(r"_instance", code_snippet):
        anti_patterns.append("过度使用单例模式")

# 2. 不必要的抽象工厂
if re.search(r"Abstract.*Factory", code_snippet) and \
       re.search(r"create.*product.*family", code_snippet, re.IGNORECASE):
# 检查是否真的需要产品族
        anti_patterns.append("可能不必要的抽象工厂")

# 3. 过深的装饰器链
    decorator_count = code_snippet.count("@")
if decorator_count > 3:
        anti_patterns.append(f"过深的装饰器链（{decorator_count}层）")

return anti_patterns

# 设计模式决策记录 (ADR)

## ADR-001: 订单系统中使用策略模式

### 状态
已采纳

### 背景
订单支付支持多种方式（支付宝、微信、信用卡），
每次新增支付方式需要修改核心业务逻辑。

### 决策
采用策略模式封装支付逻辑。

### 理由
1. 符合开闭原则：新增支付方式不修改现有代码
2. 提高可测试性：每种支付策略可独立测试
3. 符合团队技术栈：Python函数作为策略实现简单

### 后果
**正面：**
- 支付逻辑解耦
- 代码更易维护
- 新成员容易理解

**负面：**
- 增加了少量抽象层
- 需要团队成员理解策略模式

### 实施示例
```python
def alipay_pay(amount):
    pass

def wechat_pay(amount):
    pass

payment_strategies = {
    'alipay': alipay_pay,
    'wechat': wechat_pay
}

#### 5.3.2 设计模式学习资源

```python
# design_patterns/__init__.py
"""
Python设计模式最佳实践库

包含内容：
1. 常用设计模式的Pythonic实现
2. 测试示例和模式
3. 代码审查清单
4. 性能优化建议

使用指南：
1. 先阅读相应模式的理论说明
2. 查看实现示例
3. 参考测试用例
4. 应用代码审查清单

贡献指南：
1. 新增模式前确认是否已有类似实现
2. 提供充分的测试用例
3. 包含性能基准测试（如需要）
4. 更新相关文档
"""

# 阶段1：简单实现
defprocess_order(order_data):
# 直接处理
    save_to_db(order_data)
    send_email(order_data["email"], "订单确认")
    update_inventory(order_data["items"])
returnTrue

# 阶段2：函数分离
defprocess_order(order_data):
    save_order(order_data)
    notify_customer(order_data)
    update_stock(order_data)
returnTrue

defsave_order(order_data):
pass

defnotify_customer(order_data):
pass

defupdate_stock(order_data):
pass

# 阶段3：引入策略模式（如果需要）
classOrderProcessor:
def__init__(self, notification_strategy, inventory_strategy):
self.notify = notification_strategy
self.update_stock = inventory_strategy

defprocess(self, order_data):
self.save(order_data)
self.notify(order_data)
self.update_stock(order_data)
returnTrue

defsave(self, order_data):
pass

# 阶段4：引入观察者模式（如果需要）
classOrderEventSystem:
def__init__(self):
self._observers = []

defprocess_order(self, order_data):
self.save(order_data)
self._notify("order.created", order_data)
returnTrue

def_notify(self, event_type, data):
for observer inself._observers:
            observer(event_type, data)

classRefactoringChecklist:
"""重构设计模式检查清单"""

    @staticmethod
defbefore_refactoring(code_snippet):
"""重构前检查"""
        checks = {
"test_coverage": "重构前是否有足够的测试覆盖率？",
"backup": "代码是否已备份或提交到版本控制？",
"understanding": "是否充分理解现有代码逻辑？",
"impact": "是否评估了重构对系统其他部分的影响？"
        }
return checks

    @staticmethod
defduring_refactoring():
"""重构过程中"""
        principles = [
"小步前进，频繁测试",
"保持功能不变",
"每次提交一个小的变化",
"使用版本控制分支"
        ]
return principles

    @staticmethod
defafter_refactoring():
"""重构后验证"""
        validations = [
"所有测试用例通过",
"性能没有明显下降",
"代码审查通过",
"文档已更新"
        ]
return validations

1. 1. 合适性原则：选择最适合当前场景的模式，而不是最复杂的
2. 2. 渐进式演进：从简单开始，根据需要逐步引入模式
3. 3. Pythonic优先：充分利用Python语言特性，避免生搬硬套
4. 4. 可测试性设计：让代码易于测试，便于持续改进
5. 5. 团队协作规范：建立统一的命名、文档和审查标准

classDesignPatternReview:
"""设计模式使用回顾"""

def__init__(self, codebase_path):
self.path = codebase_path

defcollect_pattern_usage(self):
"""收集模式使用情况"""
        patterns = {
"factory": self._detect_factory_pattern,
"strategy": self._detect_strategy_pattern,
"observer": self._detect_observer_pattern,
"decorator": self._detect_decorator_pattern
        }

        usage_stats = {}
for pattern_name, detector in patterns.items():
            count = detector()
            usage_stats[pattern_name] = count

return usage_stats

defgenerate_recommendations(self, usage_stats):
"""生成改进建议"""
        recommendations = []

# 检查过度使用
        total_patterns = sum(usage_stats.values())
if total_patterns > 50:  # 假设阈值
            recommendations.append("考虑简化设计，减少模式使用")

# 检查特定模式使用
if usage_stats.get("singleton", 0) > 10:
            recommendations.append("单例模式使用过多，考虑其他方案")

return recommendations

import time
import statistics
from functools import wraps

classPerformanceBenchmark:
"""设计模式性能基准测试"""

def__init__(self):
self.results = {}

defbenchmark(self, func, *args, iterations=1000, **kwargs):
"""运行基准测试"""
        times = []

for _ inrange(iterations):
            start = time.perf_counter()
            func(*args, **kwargs)
            end = time.perf_counter()
            times.append(end - start)

        stats = {
"min": min(times) * 1000,  # 毫秒
"max": max(times) * 1000,
"mean": statistics.mean(times) * 1000,
"median": statistics.median(times) * 1000,
"stddev": statistics.stdev(times) * 1000iflen(times) > 1else0
        }

return stats

defcompare_implementations(self, implementations):
"""比较不同实现"""
        comparison = {}

for name, (func, args, kwargs) in implementations.items():
            stats = self.benchmark(func, *args, **kwargs)
            comparison[name] = stats

return comparison

# 使用示例
deftraditional_strategy():
# 传统策略模式实现
pass

defpythonic_strategy():
# Pythonic策略模式实现
pass

benchmark = PerformanceBenchmark()
results = benchmark.compare_implementations({
"traditional": (traditional_strategy, [], {}),
"pythonic": (pythonic_strategy, [], {})
})