Python设计模式实践

设计模式是解决常见软件设计问题的可复用方案，Python有其独特的实现方式。

创建型模式

单例模式

# 方式1：类变量实现
class Singleton:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

s1 = Singleton()
s2 = Singleton()
print(s1 is s2)  # True

# 方式2：元类实现
class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Database(metaclass=SingletonMeta):
    def __init__(self, connection_string):
        self.connection_string = connection_string

# 方式3：模块级单例（Python推荐）
# singleton.py
class _Singleton:
    def __init__(self):
        self.value = None

instance = _Singleton()

# 使用
from singleton import instance

工厂模式

# 简单工厂
class AnimalFactory:
    @staticmethod
    def create(animal_type):
        animals = {
            'dog': Dog,
            'cat': Cat,
            'bird': Bird
        }
        return animals.get(animal_type)()

class Dog:
    def speak(self):
        return "Woof"

class Cat:
    def speak(self):
        return "Meow"

animal = AnimalFactory.create('dog')

# 抽象工厂
class AnimalFactory(ABC):
    @abstractmethod
    def create_pet(self):
        pass

class DogFactory(AnimalFactory):
    def create_pet(self):
        return Dog()

class CatFactory(AnimalFactory):
    def create_pet(self):
        return Cat()

# 客户端
factory = DogFactory()
pet = factory.create_pet()

建造者模式

class ComputerBuilder:
    def __init__(self):
        self._computer = Computer()

    def set_cpu(self, cpu):
        self._computer.cpu = cpu
        return self

    def set_ram(self, ram):
        self._computer.ram = ram
        return self

    def set_storage(self, storage):
        self._computer.storage = storage
        return self

    def build(self):
        return self._computer

class Computer:
    def __init__(self):
        self.cpu = None
        self.ram = None
        self.storage = None

    def __str__(self):
        return f"CPU: {self.cpu}, RAM: {self.ram}, Storage: {self.storage}"

# 使用链式调用
computer = ComputerBuilder() \
    .set_cpu("Intel i7") \
    .set_ram("16GB") \
    .set_storage("512GB SSD") \
    .build()

结构型模式

适配器模式

# 目标接口
class Target:
    def request(self):
        return "Target request"

# 被适配类
class Adaptee:
    def specific_request(self):
        return "Adaptee specific request"

# 适配器
class Adapter(Target):
    def __init__(self, adaptee):
        self.adaptee = adaptee

    def request(self):
        return self.adaptee.specific_request()

adaptee = Adaptee()
adapter = Adapter(adaptee)
adapter.request()

装饰器模式

# Python原生装饰器实现

def logging_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"Finished {func.__name__}")
        return result
    return wrapper

def retry_decorator(max_retries=3):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == max_retries - 1:
                        raise
                    print(f"Retry {i+1}")
        return wrapper
    return decorator

@logging_decorator
@retry_decorator(max_retries=3)
def fetch_data(url):
    return requests.get(url)

# 类装饰器
class CountDecorator:
    def __init__(self, func):
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"Call count: {self.count}")
        return self.func(*args, **kwargs)

@CountDecorator
def greet(name):
    return f"Hello {name}"

代理模式

class RealService:
    def operation(self):
        return "Real operation"

class Proxy:
    def __init__(self, real_service):
        self.real_service = real_service
        self._cache = None

    def operation(self):
        # 缓存代理
        if self._cache is None:
            self._cache = self.real_service.operation()
        return self._cache

# 保护代理
class AccessProxy:
    def __init__(self, real_service, user):
        self.real_service = real_service
        self.user = user

    def operation(self):
        if self.user.role != 'admin':
            raise PermissionError("Access denied")
        return self.real_service.operation()

行为型模式

策略模式

from abc import ABC, abstractmethod

class SortStrategy(ABC):
    @abstractmethod
    def sort(self, data):
        pass

class QuickSort(SortStrategy):
    def sort(self, data):
        return sorted(data)  # Python内置快速排序

class MergeSort(SortStrategy):
    def sort(self, data):
        # 实现归并排序
        pass

class Sorter:
    def __init__(self, strategy: SortStrategy):
        self.strategy = strategy

    def sort(self, data):
        return self.strategy.sort(data)

# 动态切换策略
sorter = Sorter(QuickSort())
result = sorter.sort([3, 1, 4, 1, 5])
sorter.strategy = MergeSort()

观察者模式

class Observer:
    def update(self, data):
        pass

class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer: Observer):
        self._observers.append(observer)

    def detach(self, observer: Observer):
        self._observers.remove(observer)

    def notify(self, data):
        for observer in self._observers:
            observer.update(data)

class EmailNotifier(Observer):
    def update(self, data):
        send_email(data)

class SMSNotifier(Observer):
    def update(self, data):
        send_sms(data)

# 使用
subject = Subject()
subject.attach(EmailNotifier())
subject.attach(SMSNotifier())
subject.notify("New event")

命令模式

class Command(ABC):
    @abstractmethod
    def execute(self):
        pass

    @abstractmethod
    def undo(self):
        pass

class CreateFileCommand(Command):
    def __init__(self, filename):
        self.filename = filename

    def execute(self):
        open(self.filename, 'w').close()

    def undo(self):
        os.remove(self.filename)

class CommandManager:
    def __init__(self):
        self._history = []

    def execute(self, command: Command):
        command.execute()
        self._history.append(command)

    def undo(self):
        if self._history:
            command = self._history.pop()
            command.undo()

模板方法模式

from abc import ABC, abstractmethod

class DataProcessor(ABC):
    def process(self, data):
        data = self.read(data)
        data = self.validate(data)
        data = self.transform(data)
        self.save(data)

    @abstractmethod
    def read(self, source):
        pass

    def validate(self, data):  # 可选覆盖
        return data

    @abstractmethod
    def transform(self, data):
        pass

    @abstractmethod
    def save(self, data):
        pass

class CSVProcessor(DataProcessor):
    def read(self, source):
        return pd.read_csv(source)

    def transform(self, data):
        return data.dropna()

    def save(self, data):
        data.to_csv('output.csv')

class JSONProcessor(DataProcessor):
    def read(self, source):
        return json.load(source)

    def transform(self, data):
        return {k: v for k, v in data.items() if v is not None}

    def save(self, data):
        json.dump(data, open('output.json', 'w'))

状态模式

class State(ABC):
    @abstractmethod
    def handle(self, context):
        pass

class DraftState(State):
    def handle(self, context):
        context.state = ModerationState()
        return "Moved to moderation"

class ModerationState(State):
    def handle(self, context):
        context.state = PublishedState()
        return "Published"

class PublishedState(State):
    def handle(self, context):
        context.state = DraftState()
        return "Back to draft"

class Document:
    def __init__(self):
        self.state = DraftState()

    def process(self):
        return self.state.handle(self)

Python特有实现

使用字典实现策略

# 替代策略模式的简单方式
STRATEGIES = {
    'quick': lambda data: sorted(data),
    'merge': merge_sort,
    'bubble': bubble_sort
}

def sort_data(data, strategy='quick'):
    return STRATEGIES[strategy](data)

使用函数替代命令

# 函数作为命令
class UndoManager:
    def __init__(self):
        self._undo_stack = []

    def do(self, action, undo_action):
        action()
        self._undo_stack.append(undo_action)

    def undo(self):
        if self._undo_stack:
            self._undo_stack.pop()()

manager = UndoManager()
manager.do(
    lambda: create_file('test.txt'),
    lambda: delete_file('test.txt')
)

使用contextlib实现模板

from contextlib import contextmanager

@contextmanager
def database_transaction():
    db = Database()
    db.begin_transaction()
    try:
        yield db
        db.commit()
    except:
        db.rollback()
        raise

with database_transaction() as db:
    db.execute("INSERT ...")

设计模式选择指南

注意：Python的动态特性使得很多模式可以用更简单的方式实现，不要过度设计。

要点总结

• 单例：Python推荐模块级单例或元类实现
• 工厂：使用字典映射简化简单工厂，抽象工厂用于系列创建
• 装饰器：Python原生支持@decorator语法，支持函数和类装饰器
• 策略：策略类或函数字典，动态切换算法实现
• 观察者：Subject维护观察者列表，notify触发更新
• 命令：封装操作为对象，支持撤销和队列
• 根据问题类型选择模式，Python动态特性可简化实现

存放路径：articles/PYTHON/专家/架构与设计/设计模式在Python.md