Python 设计模式与 SOLID：从经典模式到 Pythonic 简化

from typing import Protocol

# ISP：用 Protocol 定义最小接口
class Notifier(Protocol):
    def send(self, message: str) -> None: ...

# DIP：高层模块依赖抽象 Notifier，不依赖具体 EmailSms
class OrderService:
    def __init__(self, notifier: Notifier):
        self._notifier = notifier   # 依赖注入

class EmailNotifier:
    def send(self, message: str) -> None:
        print(f"[email] {message}")

class SmsNotifier:
    def send(self, message: str) -> None:
        print(f"[sms] {message}")

service = OrderService(EmailNotifier())
service2 = OrderService(SmsNotifier())  # 开闭：新增通知方式不改 OrderService

# 方式一：模块级对象（最 Pythonic，且天然线程安全）
# db.py
_connection = None

def get_connection():
    global _connection
    if _connection is None:
        _connection = connect_to_db()
    return _connection

# 方式二：__new__ 控制
class Singleton:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

# 方式三：装饰器
def singleton(cls):
    instances = {}
    def get(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get

@singleton
class Config: ...

# 简单工厂：根据类型创建对象
class Animal:
    def speak(self): raise NotImplementedError

class Dog(Animal):
    def speak(self): return "汪汪"

class Cat(Animal):
    def speak(self): return "喵喵"

class AnimalFactory:
    _registry = {"dog": Dog, "cat": Cat}
    @classmethod
    def create(cls, kind: str) -> Animal:
        try:
            return cls._registry[kind]()
        except KeyError:
            raise ValueError(f"未知动物: {kind}")

# 建造者：链式设置可选参数
class Pizza:
    def __init__(self):
        self.size = None; self.toppings = []
    def __repr__(self): return f"Pizza(size={self.size}, toppings={self.toppings})"

class PizzaBuilder:
    def __init__(self): self._pizza = Pizza()
    def size(self, s): self._pizza.size = s; return self
    def add_topping(self, t): self._pizza.toppings.append(t); return self
    def build(self): return self._pizza

pizza = PizzaBuilder().size("L").add_topping("cheese").add_topping("mushroom").build()

# 让不兼容接口协同工作
class OldPayment:
    def pay(self, amount): return f"Old pay {amount}"

class NewPayment:
    def charge(self, cents): return f"New charge {cents}"

class NewPaymentAdapter:
    """把 NewPayment 适配成旧接口"""
    def __init__(self, new: NewPayment): self._new = new
    def pay(self, amount): return self._new.charge(int(amount * 100))

def checkout(payment):
    print(payment.pay(99.9))

checkout(OldPayment())
checkout(NewPaymentAdapter(NewPayment()))

import functools, time

def retry(max_attempts=3, delay=1.0):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f"重试 {attempt+1}: {e}")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3)
def fetch_data():
    ...

# 代理：延迟加载 + 权限控制
class LazyFile:
    def __init__(self, path): self.path = path; self._content = None
    @property
    def content(self):
        if self._content is None:
            self._content = open(self.path, encoding="utf-8").read()
        return self._content

class EventEmitter:
    def __init__(self):
        self._listeners = {}
    def on(self, event, fn):
        self._listeners.setdefault(event, []).append(fn)
        return self
    def emit(self, event, *args):
        for fn in self._listeners.get(event, []):
            fn(*args)

# 使用
emitter = EventEmitter()
emitter.on("login", lambda user: print(f"通知用户 {user} 登录"))
emitter.on("login", log_security_event)  # 多个订阅者
emitter.emit("login", "alice")

# 策略：注入可替换算法（用函数即可，不必类）
def bubble_sort(data): return sorted(data)
def quick_sort(data): return sorted(data)  # 示意

class Sorter:
    def __init__(self, strategy=None):
        self._strategy = strategy or bubble_sort
    def sort(self, data):
        return self._strategy(data)

# 模板方法：父类定骨架，子类填实现
class DataPipeline:
    def run(self):
        data = self.load()          # 子类实现
        cleaned = self.clean(data)  # 子类实现
        self.save(cleaned)          # 子类实现
        self.report(len(cleaned))   # 骨架（默认实现）

class CsvPipeline(DataPipeline):
    def load(self): ...
    def clean(self, data): ...
    def save(self, cleaned): ...

from functools import singledispatch

# Visitor → singledispatch（多态分派）
@singledispatch
def area(shape):
    raise NotImplementedError(f"不支持 {type(shape)}")

class Circle: ...
class Rect: ...

@area.register
def _(shape: Circle):
    return 3.14 * shape.radius ** 2

@area.register
def _(shape: Rect):
    return shape.w * shape.h

# 生成器替代迭代器模式
def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

from typing import Iterable, Callable, TypeVar
T = TypeVar("T")

# 管道模式：数据经一串纯函数变换
def pipe(data: T, *fns: Callable[[T], T]) -> T:
    for fn in fns:
        data = fn(data)
    return data

# 组合
def compose(*fns):
    def _apply(x):
        for fn in reversed(fns):
            x = fn(x)
        return x
    return _apply

# 使用
result = pipe(
    read_csv("data.csv"),
    filter_rows(lambda r: r["age"] > 18),
    sort_by("salary", reverse=True),
    take(10),
)

from typing import Protocol
from dataclasses import dataclass

# --- 领域层（DIP：依赖抽象）---
class PaymentGateway(Protocol):
    def charge(self, amount_cents: int) -> str: ...

@dataclass
class Order:
    amount: int
    currency: str = "CNY"

# --- 具体实现（可插拔）---
class StripeGateway:
    def charge(self, amount_cents: int) -> str:
        return f"Stripe charged {amount_cents}"

class AlipayGateway:
    def charge(self, amount_cents: int) -> str:
        return f"Alipay charged {amount_cents}"

# --- 工厂：按配置选择实现 ---
GATEWAYS = {
    "stripe": StripeGateway,
    "alipay": AlipayGateway,
}

class GatewayFactory:
    @classmethod
    def create(cls, name: str) -> PaymentGateway:
        try:
            return GATEWAYS[name]()
        except KeyError:
            raise ValueError(f"不支持的支付渠道: {name}")

# --- 服务：依赖注入 + 开闭 ---
class PaymentService:
    def __init__(self, gateway: PaymentGateway):
        self._gateway = gateway      # 构造注入

    def pay(self, order: Order) -> str:
        return self._gateway.charge(order.amount * 100)

# --- 装配 ---
service = PaymentService(GatewayFactory.create("alipay"))
print(service.pay(Order(99.9)))     # Alipay charged 9990