实战案例

# 综合：订单处理流水线
import functools, time

def logged(func):                    # 装饰器：日志
    @functools.wraps(func)
    def wrapper(*a, **kw):
        print(f"[{time.strftime('%H:%M:%S')}] 执行 {func.__name__}")
        return func(*a, **kw)
    return wrapper

class Notifier:                      # 观察者：通知
    def __init__(self):
        self.subs = []
    def attach(self, fn):
        self.subs.append(fn)
    def notify(self, msg):
        for fn in self.subs:
            fn(msg)

class ShippingFactory:               # 工厂：运费策略
    @staticmethod
    def create(kind):
        return {"standard": lambda w: w * 5,
                "express": lambda w: w * 12}[kind]

class OrderSystem:
    def __init__(self):
        self.notifier = Notifier()

    @logged
    def place_order(self, weight, shipping):
        ship = ShippingFactory.create(shipping)
        cost = ship(weight)
        self.notifier.notify(f"订单已创建，运费 {cost} 元")
        return cost

# 使用
sys = OrderSystem()
sys.notifier.attach(lambda m: print("[邮件]", m))
sys.notifier.attach(lambda m: print("[短信]", m))
sys.place_order(3, "express")