进阶篇 / 第 12 章

第 12 章 · 面向对象进阶与设计模式

为什么要「进阶」：从会用到用好

很多初学者写完类之后就不再碰面向对象（Object-Oriented Programming，OOP），代码永远是「一个类、一个 __init__、几个方法」。这种代码在小脚本里够用，一旦超过 500 行就会暴露出一堆问题：状态混乱、方法越加越长、测试时必须把整个对象构造出来、想复用某个功能只能整类继承……

面向对象真正的价值不在于「把代码塞进盒子里」，而在于三件事：

1. 封装（encapsulation）：把数据与操作数据的方法绑定在一起，对外只暴露稳定接口。
2. 多态（polymorphism）：让不同类型的对象对同一条消息做出不同响应，从而让调用方不必写 if isinstance(...)。
3. 解耦（decoupling）：让「谁和谁交互」变成可替换的选择，而不是写死的依赖。

设计模式（design pattern）就是这三件事的套路化沉淀。1994 年四人帮（GoF）提出 23 个模式，其中在 Python 里真正高估的只有少数几个。下面先看一段「初学者写法」与「进阶写法」的对照，建立感觉：

初学者写法：分支地狱

# order.py —— 用 if/elif 区分不同类型的订单
class Order:
    def __init__(self, kind, amount, weight, volume):
        self.kind = kind
        self.amount = amount
        self.weight = weight      # 重量订单才有意义
        self.volume = volume      # 体积订单才有意义

    def calc_fee(self):
        if self.kind == "normal":
            fee = 10
        elif self.kind == "weight":
            fee = 5 + self.weight * 2
        elif self.kind == "volume":
            fee = 8 + self.volume * 3
        elif self.kind == "member":
            fee = self.amount * 0.9
        else:
            raise ValueError("未知订单类型")
        return fee

    def describe(self):
        # describe 里还要再写一遍同样的分支
        if self.kind == "member":
            return f"会员订单，实付 {self.amount * 0.9}"
        return f"订单，运费 {self.calc_fee()}"

o1 = Order("weight", 100, 3.2, 0)
o2 = Order("member", 200, 0, 0)
print(o1.describe())
print(o2.describe())

输出：

订单，运费 11.4
会员订单，实付 180.0

进阶写法：策略对象（本章 s8 详解）

from dataclasses import dataclass
from decimal import Decimal

class FeeStrategy:
    """费用策略接口（基类）。"""
    def calc(self, order) -> Decimal:
        raise NotImplementedError

class WeightFee(FeeStrategy):
    def __init__(self, base: Decimal = Decimal("5"), rate: Decimal = Decimal("2")):
        self.base, self.rate = base, rate
    def calc(self, order) -> Decimal:
        return self.base + order.weight * self.rate

class MemberFee(FeeStrategy):
    def __init__(self, discount: Decimal = Decimal("0.9")):
        self.discount = discount
    def calc(self, order) -> Decimal:
        return order.amount * self.discount

@dataclass
class Order:
    amount: Decimal
    weight: Decimal = Decimal("0")
    volume: Decimal = Decimal("0")
    strategy: FeeStrategy = None      # 运行时注入策略

    def __post_init__(self):
        if self.strategy is None:
            self.strategy = WeightFee()

    @property
    def fee(self) -> Decimal:
        return self.strategy.calc(self)

print(Order(Decimal("100"), weight=Decimal("3.2"), strategy=WeightFee()).fee)
print(Order(Decimal("200"), strategy=MemberFee()).fee)

输出：

11.4
180.0

property：把方法伪装成属性

如果你来自 Java 或 C#，你会习惯写 getXxx() / setXxx()。Python 不需要：@property 装饰器（decorator）让你可以写 obj.name 而背后仍然是方法调用。这不只是语法糖，它解决了真实痛点——公开字段可以在不破坏调用方的前提下变成受控属性。

account.py

from __future__ import annotations

class Account:
    def __init__(self, owner: str, balance: float = 0.0) -> None:
        self.owner = owner          # 公开属性：直接读写
        self._balance = balance    # 私有属性：约定不要外部直接碰

    # 只读属性：外部只能读，想写就报错
    @property
    def balance(self) -> float:
        """当前余额（只读）。"""
        return self._balance

    # 可写属性：赋值时自动走校验逻辑
    @balance.setter
    def balance(self, value: float) -> None:
        if value < 0:
            raise ValueError("余额不能为负")
        self._balance = value

    # 计算属性：不需要存储，每次现算
    @property
    def level(self) -> str:
        b = self._balance
        if b >= 10000:
            return "金卡"
        if b >= 1000:
            return "银卡"
        return "普通"

a = Account("张三", 200)
print(a.balance)          # 200.0     像字段一样读
print(a.level)            # 普通      像字段一样读
a.balance = 5000
print(a.level)            # 银卡      值变了，派生属性自动跟着变
a.balance = -1            # ValueError: 余额不能为负

缓存属性：functools.cached_property

如果某个 property 计算很贵（比如要读数据库、算一次复杂报表），你可能想「算一次存起来」。Python 3.8+ 提供 functools.cached_property，它把结果写进实例的 __dict__，后续访问直接命中缓存：

cached_demo.py

from functools import cached_property
import time

class Report:
    def __init__(self, rows: list[str]) -> None:
        self.rows = rows
        self.call_count = 0

    @cached_property
    def stats(self) -> dict:
        # 假装很慢：解析、聚合
        self.call_count += 1
        time.sleep(0.2)
        nums = [float(r) for r in self.rows]
        return {"sum": sum(nums), "avg": sum(nums) / len(nums)}

r = Report(["1", "2", "3", "4"])
t0 = time.perf_counter()
print(r.stats)      # 触发计算
print(r.stats)      # 命中缓存
print(r.stats["sum"], "计算次数 =", r.call_count)
print(f"耗时 {time.perf_counter() - t0:.2f}s")

输出：

{'sum': 10.0, 'avg': 2.5}
{'sum': 10.0, 'avg': 2.5}
10.0 计算次数 = 1
耗时 0.20s

属性级校验：setter 还是 __setattr__

当很多字段都要同一种校验时，逐个写 setter 很啰嗦。这时可以重写 __setattr__（Python 层面 Python 类唯一无法用 property 拦截的赋值入口）：

validate.py

class PositiveModel:
    """所有以 _ 开头的属性赋值时必须 > 0。"""
    def __setattr__(self, name, value):
        if name.startswith("_"):
            object.__setattr__(self, name, value)
        elif not isinstance(value, (int, float)) or value <= 0:
            raise ValueError(f"{name} 必须是正数，得到 {value!r}")
        object.__setattr__(self, name, value)

class Product(PositiveModel):
    def __init__(self, price: float, stock: int) -> None:
        self.price = price
        self.stock = stock

p = Product(9.9, 100)
print(p.price, p.stock)
p.stock = 50
print(p.stock)
try:
    p.price = -1
except ValueError as e:
    print("校验失败:", e)

输出：

9.9 100
50
校验失败: price 必须是正数，得到 -1

混入类（Mixin）与方法解析顺序 MRO

Python 的类可以继承多个类，最右侧的会先被初始化。混入类（mixin）是一个不打算被直接实例化、只提供一组可复用方法的类，它通常没有 __init__，也不维护自己的状态——它只往混入结果里「加能力」。

mixin.py

from datetime import datetime

class TimestampMixin:
    """给任何类加上 创建时间 / 最近更新时间。"""
    _CREATED_KEY = "_created_at"
    _UPDATED_KEY = "_updated_at"

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        now = datetime.now()
        setattr(self, self._CREATED_KEY, now)
        setattr(self, self._UPDATED_KEY, now)

    def touch(self) -> None:
        setattr(self, self._UPDATED_KEY, datetime.now())

    @property
    def age_seconds(self) -> float:
        return (datetime.now() - getattr(self, self._UPDATED_KEY)).total_seconds()

class SerializableMixin:
    def to_dict(self) -> dict:
        return {k: v for k, v in vars(self).items() if not k.startswith("_")}

class User(TimestampMixin, SerializableMixin):
    def __init__(self, name: str, age: int = 0) -> None:
        self.name, self.age = name, age
        super().__init__()          # 关键：手动把控制权交给 MRO 的下一站

u = User("李四", 30)
print(u.name, u.age, u.to_dict())
u.name = "李四改"
u.touch()
print(f"修改后存活 {u.age_seconds:.4f} 秒")

输出：

李四 30 {'name': '李四', 'age': 30}
修改后存活 0.0000 秒

MRO：方法解析顺序（Method Resolution Order）

多继承时 Python 用 C3 线性化算法（C3 linearization）算出一条确定的查找顺序，可以用 cls.__mro__ 或 cls.mro() 查看：

mro.py

class A:
    def who(self) -> str: return "A"

class B(A):
    def who(self) -> str: return "B -> " + super().who()

class C(A):
    def who(self) -> str: return "C -> " + super().who()

class D(B, C):
    def who(self) -> str: return "D -> " + super().who()

print(" -> ".join(c.__name__ for c in D.__mro__))
print(D().who())

class X(A): pass
class Y(A): pass
try:
    class Z(X, Y): pass
except TypeError as e:
    print("创建失败:", e)

输出：

D -> B -> C -> A -> object
D -> B -> C -> A
创建失败: Cannot create a consistent method resolution order (MRO) for bases X, Y

组合优于继承：继承树 vs 对象图

Java 教我们「is-a 关系用继承」，Go 教我们「优先组合（composition）优于继承（composition over inheritance）」。Python 里这条原则更被强化，因为 Python 的类可以运行时修改、可以被 monkey patch、多继承成本极低——一旦你依赖继承，调用方就再也无法替换你的实现。

反例：用继承表达「有一个」

bad_inheritance.py

class FileLogger:
    def __init__(self, filename):
        self.filename = filename
    def log(self, msg):
        with open(self.filename, "a", encoding="utf-8") as f:
            f.write(msg + "\n")

# 错误：Service「有一个」logger，不是「是一种」FileLogger
class UserService(FileLogger):
    def __init__(self, filename):
        super().__init__(filename)

svc = UserService("/tmp/svc.log")
svc.log("服务启动")     # 能跑，但它不能换成打印到控制台
svc.filename = None     # 调用方一不小心就把它玩坏了

正例：组合 + 依赖注入

good_composition.py

from typing import Protocol
from datetime import datetime

class Logger(Protocol):
    def log(self, msg: str) -> None: ...

class FileLogger:
    def __init__(self, filename: str) -> None:
        self.filename = filename
    def log(self, msg: str) -> None:
        with open(self.filename, "a", encoding="utf-8") as f:
            f.write(f"{datetime.now():%H:%M:%S} {msg}\n")
    def __repr__(self) -> str:
        return f"FileLogger({self.filename!r})"

class ConsoleLogger:
    def log(self, msg: str) -> None:
        print(f"[LOG] {msg}")
    def __repr__(self) -> str:
        return "ConsoleLogger()"

class MemoryLogger:
    def __init__(self) -> None:
        self.records: list[str] = []
    def log(self, msg: str) -> None:
        self.records.append(msg)

class UserService:
    def __init__(self, logger: Logger) -> None:
        self.logger = logger        # 组合：持有别人，而不是变成别人
    def register(self, name: str) -> None:
        if not name:
            self.logger.log("注册失败：姓名为空")
            return
        self.logger.log(f"用户注册成功：{name}")

svc = UserService(ConsoleLogger())
svc.register("张三")
mem = MemoryLogger()
UserService(mem).register("李四")
print(mem.records)
print(UserService(FileLogger("/tmp/app.log")).logger)

输出：

[LOG] 用户注册成功：张三
['用户注册成功：李四']
FileLogger('/tmp/app.log')

单例模式：全局唯一实例

单例模式（Singleton Pattern）保证一个类在进程内只有一个实例，典型用途：配置管理器、连接池、全局缓存、硬件句柄（如 GPU、数据库连接）。Python 里实现方式比 Java 简单得多，甚至不需要模式本身——模块本身就是单例。

方式一：模块级（Pythonic，最推荐）

# config.py —— 模块天然只会被执行一次
import os

class Config:
    def __init__(self) -> None:
        self.debug = os.environ.get("DEBUG", "0") == "1"

config = Config()      # 全局唯一实例

# main.py
from config import config
print(config.debug)

方式二：__new__ 经典实现

import threading

class Singleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        # 双重检查：先无锁快速路径，再加锁慢路径
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, value: int = 0) -> None:
        # 注意：__init__ 每次 __new__ 返回已存在实例时仍会被调用
        if not hasattr(self, "_ready"):
            self.value = value
            self._ready = True

a, b = Singleton(1), Singleton(2)
print(a is b, a.value, b.value)      # True 1 1  —— b 的 __init__ 没有覆盖值

输出：

True 1 1

方式三：装饰器（最通用）

import functools

def singleton(cls):
    """把任意类包装成单例，保留原类名与文档。"""
    @functools.wraps(cls)
    def wrapper(*args, **kwargs):
        if not hasattr(wrapper, "_instance"):
            wrapper._instance = cls(*args, **kwargs)
        return wrapper._instance
    wrapper._instance = None
    return wrapper

@singleton
class AppState:
    def __init__(self) -> None:
        self.items: list[str] = []
    def add(self, s: str) -> None:
        self.items.append(s)

s1, s2 = AppState(), AppState()
s1.add("hello")
print(s1 is s2, s2.items, AppState.__name__)

输出：

True ['hello'] AppState

方式四：枚举类（现代 Python 的最佳答案）

from enum import Enum

class AppState(Enum):
    """枚举天然单例，成员身份唯一、不可变、可比较。"""
    RUNNING = "running"
    PAUSED = "paused"
    STOPPED = "stopped"

    @property
    def is_active(self) -> bool:
        return self is AppState.RUNNING

print(AppState.RUNNING is AppState.RUNNING)      # True
print(AppState("running"))                        # 按值反查
print(AppState.RUNNING.is_active)                 # True
print(list(AppState))                              # 遍历成员

输出：

True
<AppState.RUNNING: 'running'>
True
[<AppState.RUNNING: 'running'>, <AppState.PAUSED: 'paused'>, <AppState.STOPPED: 'stopped'>]

工厂模式：把「创建谁」的知识封装起来

工厂模式（Factory Pattern）解决一个具体问题：调用方不应该知道要创建哪一个具体类。它把实例化逻辑集中到一个地方，将来增删实现时只改一处。

6.1 简单工厂函数

from dataclasses import dataclass

@dataclass
class Circle:
    radius: float

@dataclass
class Square:
    side: float

def shape_factory(kind: str, size: float):
    """根据名字创建图形；未知类型给出清晰错误。"""
    table = {
        "circle": Circle,
        "square": Square,
    }
    cls = table.get(kind)
    if cls is None:
        raise ValueError(f"未知图形: {kind}，可选 {sorted(table)}")
    return cls(size)

for k in ("circle", "square"):
    print(shape_factory(k, 3))
try:
    shape_factory("triangle", 3)
except ValueError as e:
    print(e)

输出：

Circle(radius=3.0)
Square(side=3.0)
未知图形: triangle，可选 ['circle', 'square']

6.2 注册式工厂：新增类型不改旧代码

这是插件系统、命令行工具、序列化框架的核心套路：每个实现模块自己注册自己，工厂只负责查表。

registry.py

from typing import Any, Callable

_REGISTRY: dict[str, Callable[..., Any]] = {}

def register(name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """装饰器：把一个类注册到工厂表中。"""
    def deco(cls):
        if name in _REGISTRY:
            raise ValueError(f"{name} 已注册为 {_REGISTRY[name]}")
        _REGISTRY[name] = cls
        return cls
    return deco

def create(name: str, **kwargs) -> Any:
    if name not in _REGISTRY:
        raise KeyError(f"未注册的处理器 {name}，可用: {sorted(_REGISTRY)}")
    return _REGISTRY[name](**kwargs)

def available() -> list[str]:
    return sorted(_REGISTRY)

handlers.py

from registry import register, create

@register("json")
class JsonHandler:
    def __init__(self, *, indent: int = 2) -> None:
        self.indent = indent
    def dump(self, obj) -> str:
        import json
        return json.dumps(obj, indent=self.indent, ensure_ascii=False)

@register("csv")
class CsvHandler:
    def __init__(self, *, delimiter: str = ",") -> None:
        self.delimiter = delimiter
    def dump(self, obj: dict) -> str:
        return self.delimiter.join(str(v) for v in obj.values())

use_factory.py

import handlers            # 触发注册
from registry import create, available

print(available())
h = create("json", indent=0)
print(h.dump({"名称": "张三", "年龄": 18}))
print(create("csv", delimiter=" | ").dump({"a": 1, "b": 2}))
try:
    create("xml")
except KeyError as e:
    print("异常:", e)

输出：

['csv', 'json']
{"名称": "张三", "年龄": 18}
1 | 2
异常: "'未注册的处理器 xml，可用: [\'csv\', \'json\']'"

6.3 抽象工厂：一个工厂造一整族对象

当产品有「家族关系」（GU主题：按钮 + 输入框必须同主题）时，需要抽象工厂：

gui.py

from abc import ABC, abstractmethod

# —— 产品 ——
class Button(ABC):
    @abstractmethod
    def render(self) -> str: ...

class TextBox(ABC):
    @abstractmethod
    def render(self) -> str: ...

# —— 具体产品 ——
class DarkButton(Button):
    def render(self) -> str: return "[ 黑色按钮 ]"

class DarkTextBox(TextBox):
    def render(self) -> str: return "[ 黑色输入框 ]"

class LightButton(Button):
    def render(self) -> str: return "( 浅色按钮 )"

class LightTextBox(TextBox):
    def render(self) -> str: return "( 浅色输入框 )"

# —— 抽象工厂 ——
class UIFactory(ABC):
    @abstractmethod
    def make_button(self) -> Button: ...
    @abstractmethod
    def make_textbox(self) -> TextBox: ...

# —— 具体工厂 ——
class DarkFactory(UIFactory):
    def make_button(self) -> Button: return DarkButton()
    def make_textbox(self) ->TextBox: return DarkTextBox()

class LightFactory(UIFactory):
    def make_button(self) -> Button: return LightButton()
    def make_textbox(self) -> TextBox: return LightTextBox()

def render_app(factory: UIFactory) -> None:
    # 客户端代码完全不知道具体产品类
    print(factory.make_button().render())
    print(factory.make_textbox().render())

render_app(DarkFactory())
render_app(LightFactory())

输出：

[ 黑色按钮 ]
[ 黑色输入框 ]
( 浅色按钮 )
( 浅色输入框 )

6.4 工厂方法与 Python 的对应物

观察者模式：让对象状态变化自动通知他人

观察者模式（Observer Pattern，也叫发布-订阅 / publish-subscribe）定义了一对多依赖：一个主题（subject）状态变化时，所有订阅者（observer）自动收到通知。GUI 事件、消息队列、日志、邮件提醒、Webhook 全是这个模式。

7.1 朴素实现：用可调用对象当观察者

from dataclasses import dataclass, field
from typing import Callable

@dataclass
class EventBus:
    """极简事件总线：主题持有订阅者列表。"""
    _subscribers: dict[str, list[Callable]] = field(default_factory=dict)

    def subscribe(self, event: str, fn: Callable) -> Callable:
        self._subscribers.setdefault(event, []).append(fn)
        return fn                      # 返回它本身，方便当装饰器用

    def unsubscribe(self, event: str, fn: Callable) -> None:
        self._subscribers.get(event, []).remove(fn)

    def emit(self, event: str, **payload) -> None:
        for fn in list(self._subscribers.get(event, [])):   # 复制一份，允许回调中退订
            try:
                fn(**payload)
            except Exception as e:                          # 一个坏订阅者不拖垮全场
                print(f"[warn] 订阅者 {getattr(fn, '__name__', fn)} 出错: {e}")

bus = EventBus()

@bus.subscribe("order.created")          # 订阅函数直接返回自身 -> 可作装饰器
def log_order(order_id: int, amount: float) -> None:
    print(f"[log] 订单 {order_id} 创建，金额 {amount}")

@bus.subscribe("order.created")
def send_sms(order_id: int, amount: float) -> None:
    print(f"[sms] 给订单 {order_id} 发短信提醒")

bus.emit("order.created", order_id=1001, amount=99.5)
bus.emit("order.paid", order_id=1001)     # 没人订阅，什么也不发生

输出：

[log] 订单 1001 创建，金额 99.5
[sms] 给订单 1001 发短信提醒

7.2 弱引用观察者：防止内存泄漏

强引用列表会让观察者永远无法被回收（经典内存泄漏）。用 weakref 存订阅者可以避免这个问题：

weak_bus.py

import weakref
from typing import Callable

class WeakEventBus:
    def __init__(self) -> None:
        self._subs: dict[str, list[weakref.ref]] = {}

    def subscribe(self, event: str, fn: Callable) -> Callable:
        self._subs.setdefault(event, []).append(weakref.ref(fn))
        return fn

    def emit(self, event: str, **payload) -> None:
        alive = []
        for ref in list(self._subs.get(event, [])):
            fn = ref()
            if fn is None:                 # 对象已被回收，清理死引用
                continue
            alive.append(ref)
            fn(**payload)
        self._subs[event] = alive

bus = WeakEventBus()

def handler(x: int) -> None:
    print("收到", x)

bus.subscribe("tick", handler)
bus.emit("tick", x=1)

# 绑定方法订阅时需要保活，否则对象一被回收订阅就消失
class Watcher:
    def __init__(self, b: WeakEventBus) -> None:
        self._bound = self.on_tick
        b.subscribe("tick", self._bound)   # 故意制造强引用
    def on_tick(self, x: int) -> None:
        print("Watcher 收到", x)

w = Watcher(bus)
bus.emit("tick", x=2)
print("存活:", len(bus._subs["tick"]))

输出：

收到 1
Watcher 收到 2
存活: 2

7.3 状态变化即通知：Observable 基类

如果希望「字段一变就自动通知」，可以做一个带钩子的基类：

observable.py

from collections import defaultdict
from typing import Any, Callable

class Observable:
    """任意属性赋值都会触发对应属性的监听器。"""
    def __init__(self) -> None:
        object.__setattr__(self, "_listeners", defaultdict(list))
        object.__setattr__(self, "_suspended", False)

    def on_change(self, name: str, fn: Callable[[Any, Any], None]) -> None:
        self._listeners[name].append(fn)

    def __setattr__(self, name: str, value: Any) -> None:
        old = getattr(self, name, None)
        object.__setattr__(self, name, value)
        if not self._suspended and old != value:
            for fn in list(self._listeners.get(name, [])):
                fn(old, value)

class Temperature(Observable):
    def __init__(self, celsius: float = 0.0) -> None:
        super().__init__()
        object.__setattr__(self, "celsius", celsius)
    def __repr__(self) -> str:
        return f"Temperature({self.celsius}°C)"

t = Temperature(20)
t.on_change("celsius", lambda old, new: print(f"温度从 {old} 升到 {new}，开启空调"))
t.on_change("celsius", lambda old, new: print(f"告警日志：{old} -> {new}"))
t.celsius = 26      # 同值不会触发
t.celsius = 30

输出：

温度从 20 升到 30，开启空调
告警日志：20 -> 30

策略模式：把算法变成可替换对象

策略模式（Strategy Pattern）定义一系列算法，把每个算法封装成对象，并让它们可以互相替换。本章开头已经看过一个简化版本，这里展开讲三种实现层级。

8.1 级别一：传函数（最 Pythonic）

Python 里「算法」通常只是一个可调用对象，不需要非包装成类：

sort_key.py

from dataclasses import dataclass, field
from typing import Callable

@dataclass
class Student:
    name: str
    score: float
    height: float = 170.0

KeyFunc = Callable[[Student], object]

def by_score(s: Student) -> float: return s.score
def by_name(s: Student) -> str:   return s.name
def by_height(s: Student) -> float: return -s.height   # 负号 = 倒序

students = [
    Student("小明", 88, 175),
    Student("小红", 95, 160),
    Student("小刚", 72, 182),
]

def sort_students(data: list[Student], key: KeyFunc) -> list[Student]:
    return sorted(data, key=key)

print([s.name for s in sort_students(students, by_score)])     # 按分数升序
print([s.name for s in sort_students(students, by_name)])      # 按姓名
print([s.name for s in sort_students(students, by_height)])    # 按身高降序

# lambda 也能当策略
print([s.name for s in sort_students(students, key=lambda s: -s.score)])

输出：

['小刚', '小明', '小红']
['小明', '小红', '小刚']
['小刚', '小明', '小红']
['小红', '小明', '小刚']

8.2 级别二：策略对象 + 注册表

当策略本身需要配置（不是简单函数）时，把配置收进对象：

sort_strategy.py

from dataclasses import dataclass, field
from typing import Protocol

class Student(Protocol):
    name: str
    score: float
    height: float

class SortStrategy:
    name = "base"
    def key(self, item: Student) -> object:
        raise NotImplementedError
    def __call__(self, item: Student) -> object:
        return self.key(item)

@dataclass
class ScoreStrategy(SortStrategy):
    reverse: bool = False
    name: str = field(default="score")
    def key(self, item: Student) -> float:
        return -item.score if self.reverse else item.score

@dataclass
class HeightStrategy(SortStrategy):
    reverse: bool = False
    name: str = field(default="height")
    def key(self, item: Student) -> float:
        return -item.height if self.reverse else item.height

STRATEGIES: dict[str, SortStrategy] = {
    "score": ScoreStrategy(),
    "score_desc": ScoreStrategy(reverse=True),
    "height": HeightStrategy(),
    "height_desc": HeightStrategy(reverse=True),
}

@dataclass
class StudentX:
    name: str
    score: float
    height: float = 170.0

def sort_by(data: list[StudentX], strategy_name: str) -> list[str]:
    if strategy_name not in STRATEGIES:
        raise KeyError(f"未知排序策略 {strategy_name}，可用 {sorted(STRATEGIES)}")
    return sorted(data, key=STRATEGIES[strategy_name])

data = [StudentX("小明", 88, 175), StudentX("小红", 95, 160), StudentX("小刚", 72, 182)]
for n in ("score", "score_desc", "height_desc"):
    print(n, "->", [s.name for s in sort_by(data, n)])

输出：

score -> ['小刚', '小明', '小红']
score_desc -> ['小红', '小明', '小刚']
height_desc -> ['小刚', '小明', '小红']

8.3 级别三：Context 持有策略并支持热切换

from decimal import Decimal

class DiscountStrategy:
    """折扣策略接口。"""
    def apply(self, price: Decimal) -> Decimal:
        raise NotImplementedError

class NoDiscount(DiscountStrategy):
    def apply(self, price: Decimal) -> Decimal:
        return price

class PercentDiscount(DiscountStrategy):
    def __init__(self, percent: Decimal) -> None:
        if not Decimal("0") < percent <= Decimal("1"):
            raise ValueError("折扣率必须在 (0, 1] 之间")
        self.percent = percent
    def apply(self, price: Decimal) -> Decimal:
        return (price * self.percent).quantize(Decimal("0.01"))

class FixedDiscount(DiscountStrategy):
    def __init__(self, amount: Decimal) -> None:
        self.amount = amount
    def apply(self, price: Decimal) -> Decimal:
        return max(Decimal("0"), price - self.amount)

class Checkout:
    def __init__(self, strategy: DiscountStrategy | None = None) -> None:
        self.strategy: DiscountStrategy = strategy or NoDiscount()

    @property
    def final_price(self) -> Decimal:
        raise NotImplementedError

    def quote(self, price: Decimal) -> Decimal:
        return self.strategy.apply(price)

    def switch(self, strategy: DiscountStrategy) -> "Checkout":
        self.strategy = strategy
        return self

c = Checkout()
price = Decimal("128.00")
print(c.quote(price))
c.switch(PercentDiscount(Decimal("0.85")))
print(c.quote(price))
c.switch(FixedDiscount(Decimal("30")))
print(c.quote(price))
c.switch(NoDiscount())
print(c.quote(price))

输出：

128.00
108.80
98.00
128.00

dataclass：把「装数据的类」写到最短

@dataclass 是 Python 3.7 起标准库内置的类装饰器，自动生成 __init__、__repr__、__eq__。它把第 7 章里「样板代码」的问题彻底消除了。

9.1 从手写类到 dataclass

from dataclasses import dataclass, field

# 手写版：40 行只为装 3 个字段
class PointManual:
    def __init__(self, x: float, y: float, tags: list[str]) -> None:
        self.x, self.y, self.tags = x, y, tags
    def __repr__(self) -> str:
        return f"PointManual(x={self.x}, y={self.y}, tags={self.tags})"
    def __eq__(self, other: object) -> bool:
        if not isinstance(other, PointManual):
            return NotImplemented
        return (self.x, self.y, self.tags) == (other.x, other.y, other.tags)

# dataclass 版：3 行
@dataclass
class Point:
    x: float
    y: float
    tags: list[str] = field(default_factory=list)

print(Point(1, 2))
print(Point(1, 2) == PointManual(1, 2, []))     # 不同类 -> False
print(Point(1, 2) == Point(1, 2))              # 同类同值 -> True

输出：

Point(x=1, y=2, tags=[])
False
True

9.2 常用参数全表

dataclass_full.py

from dataclasses import dataclass, field, InitVar
from typing import ClassVar

@dataclass(frozen=True, slots=True, order=True)
class Version:
    major: int
    minor: int
    patch: int = 0
    def __str__(self) -> str:
        return f"v{self.major}.{self.minor}.{self.patch}"

@dataclass(slots=True, repr=False)
class User:
    name: str
    email: str
    password: str = field(repr=False, default="")     # 不打印到 repr
    login_count: int = field(default=0, compare=False) # 不参与相等比较
    KIND: ClassVar[str] = "user"                       # 类变量，不进 __init__

    def __repr__(self) -> str:
        return f"User(name={self.name!r}, email={self.email!r})"

@dataclass
class Rectangle:
    width: float
    height: float
    unit: InitVar[str] = "cm"      # 仅构造时接收，不存为属性

    def __post_init__(self, unit: str) -> None:
        if self.width <= 0 or self.height <= 0:
            raise ValueError(f"边长必须为正 ({unit})")
        self.area = self.width * self.height   # 动态添加的属性

print(Version(1, 2, 3), Version(1, 2, 3) > Version(1, 1, 9))
print(User("张三", "z@example.com", "secret").login_count)
print(User.KIND, hasattr(User("a", "b"), "KIND"))

r = Rectangle(3, 4, unit="m")
print(r.area)
try:
    Rectangle(-1, 2)
except ValueError as e:
    print("创建失败:", e)

输出：

v1.2.3 True
0
user True
12
创建失败: 边长必须为正 (cm)

9.3 dataclass 与继承

from dataclasses import dataclass, field

@dataclass
class Entity:
    id: int = field(kw_only=True)          # 基类 kw_only，子类可自由加非 kw_only 字段
    version: int = 0

@dataclass
class Product(Entity):
    name: str = ""
    price: float = 0.0
    tags: list[str] = field(default_factory=list)

p = Product(id=1, name="键盘", price=299.0, tags=["外设"])
print(p)
print(Product(1, "键盘", 299.0, ["外设"]) == p)

输出：

Product(id=1, version=0, name='键盘', price=299.0, tags=['外设'])
True

枚举 Enum：比常量类更安全的状态表示

当一个字段只允许有限个取值时，用字符串裸值会到处拼错。枚举（enum.Enum）提供「单一数据源 + 运行期校验 + 可迭代 + 可比较」。

enum_demo.py

from enum import Enum, IntEnum, StrEnum, auto, unique
from dataclasses import dataclass

@unique
class OrderStatus(Enum):
    CREATED = auto()        # auto() 自动取 1, 2, 3...
    PAID = auto()
    SHIPPED = auto()
    CANCELLED = auto()

    @property
    def is_terminal(self) -> bool:
        return self in (OrderStatus.SHIPPED, OrderStatus.CANCELLED)

    @classmethod
    def from_text(cls, text: str) -> "OrderStatus":
        try:
            return cls[text.upper()]          # 按名字查找
        except KeyError:
            raise ValueError(f"非法状态: {text}") from None

    def __str__(self) -> str:
        return self.name.lower()

class Level(IntEnum):
    LOW = 1
    HIGH = 2

class Color(StrEnum):        # Python 3.11+：值即字符串
    RED = "red"
    GREEN = "green"

@dataclass
class Order:
    no: str
    status: OrderStatus = OrderStatus.CREATED

o = Order("NO-1")
print(o.status, o.status.value, o.status.is_terminal)
o.status = OrderStatus.PAID
print(o.status, OrderStatus.from_text("shipped"))
print(Level.HIGH > Level.LOW, int(Level.HIGH), Level.HIGH == 2)
print(f"{Color.RED}", Color.RED == "red")
print([s.name for s in OrderStatus])

输出：

created 1 False
paid shipped
True 2 True
red True
['CREATED', 'PAID', 'SHIPPED', 'CANCELLED']

flag_demo.py —— 权限位组合

from enum import IntFlag, auto

class Perm(IntFlag):
    READ    = auto()      # 1
    WRITE   = auto()      # 2
    DELETE  = auto()      # 4
    EXECUTE = auto()      # 8

p = Perm.READ | Perm.WRITE
print(p, bin(p.value))
print("可读:", Perm.READ in p, "可删:", Perm.DELETE in p)
p |= Perm.DELETE
print(p)
print(Perm(7))                    # 按位数值构造
print(list(Perm.READ | Perm.WRITE))

输出：

Perm.READ|Perm.WRITE 0b11
可读: True 可删: False
Perm.READ|Perm.WRITE|Perm.DELETE
Perm.READ|Perm.WRITE|Perm.DELETE
[<Perm.READ: 1>, <Perm.WRITE: 2>]

描述符与上下文管理器等其他进阶手法

11.1 描述符（descriptor）：属性行为的可复用单元

property 是一种「数据描述符（data descriptor）」，它实现了 __get__ / __set__。你完全可以写自己的描述符，实现校验、转换、惰性缓存：

from __future__ import annotations
from typing import Any, Callable

class Positive:
    """把「必须为正」这条规则做成可复用描述符。"""
    def __set_name__(self, owner, name: str) -> None:
        self.name = "_" + name
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.name)
    def __set__(self, obj, value):
        if value <= 0:
            raise ValueError(f"{self.name} 必须为正")
        setattr(obj, self.name, value)

class Range:
    """把整数限制在 [lo, hi]。"""
    def __init__(self, lo: int, hi: int) -> None:
        self.lo, self.hi = lo, hi
    def __set_name__(self, owner, name: str) -> None:
        self.name = "_" + name
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.name, self.lo)
    def __set__(self, obj, value: int) -> None:
        if not isinstance(value, int):
            raise TypeError("必须是 int")
        value = max(self.lo, min(self.hi, value))
        setattr(obj, self.name, value)

class Temperature:
    celsius = Positive()
    humidity = Range(0, 100)

t = Temperature()
t.celsius = 36.6
t.humidity = 150          # 自动截断
print(t.celsius, t.humidity)
try:
    t.celsius = -1
except ValueError as e:
    print("校验失败:", e)
try:
    t.humidity = "high"
except TypeError as e:
    print("类型错误:", e)

输出：

36.6 100
校验失败: _celsius 必须为正
类型错误: 必须是 int

11.2 运算符重载与 __call__ 简表

from dataclasses import dataclass, field

@dataclass(frozen=True)
class Vec2:
    x: float
    y: float = 0.0

    def __add__(self, other: "Vec2") -> "Vec2":
        return Vec2(self.x + other.x, self.y + other.y)

    def __mul__(self, k: float) -> "Vec2":
        return Vec2(self.x * k, self.y * k)

    __rmul__ = __mul__          # 数字 * 向量 也走同一条路

    def __neg__(self) -> "Vec2":
        return Vec2(-self.x, -self.y)

    def __len__(self) -> int:
        return 2

    def __iter__(self):
        yield self.x
        yield self.y

    def __getitem__(self, i):
        return (self.x, self.y)[i]

a, b = Vec2(1, 2), Vec2(3, 4)
print(a + b, 3 * a, -a)
print(len(a), list(a), b[0], b[1])
print(a == Vec2(1, 2), {a: "点"})     # frozen=True -> 可哈希

输出：

Vec2(x=4.0, y=6.0) Vec2(x=3.0, y=6.0) Vec2(x=-1.0, y=-2.0)
2 [1.0, 2.0] 3.0 4.0
True {'Vec2(x=1.0, y=2.0)': '点'}

11.3 综合示例：一个订单处理流水线

把本章所有工具串起来——协议、dataclass、策略、观察者、枚举、组合：

pipeline.py

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Callable, Protocol

# 1) 策略接口（Protocol = 结构化子类型，无需继承）
class PricingRule(Protocol):
    def price(self, item: "LineItem") -> Decimal: ...

class ListPrice:
    def price(self, item: "LineItem") -> Decimal:
        return item.unit_price * item.qty

class MemberPrice:
    def price(self, item: "LineItem") -> Decimal:
        return (item.unit_price * item.qty * Decimal("0.9")).quantize(Decimal("0.01"))

# 2) 状态枚举
class OrderState(Enum):
    NEW = "new"
    PAID = "paid"
    DONE = "done"
    CANCELLED = "cancelled"

# 3) 数据模型
@dataclass
class LineItem:
    name: str
    unit_price: Decimal
    qty: int = 1
    @property
    def subtotal(self) -> Decimal:
        return self.unit_price * self.qty

# 4) 订单（组合持有策略，暴露只读派生属性）
@dataclass
class Order:
    no: str
    items: list[LineItem] = field(default_factory=list)
    rule: PricingRule = field(default_factory=ListPrice)
    state: OrderState = OrderState.NEW
    created_at: datetime = field(default_factory=datetime.now)
    _watchers: list[Callable[[str, Order], None]] = field(default_factory=list, repr=False)

    @property
    def total(self) -> Decimal:
        return sum((self.rule.price(i) for i in self.items), Decimal("0"))

    def watch(self, fn: Callable[[str, Order], None]) -> None:
        self._watchers.append(fn)

    def _notify(self, event: str) -> None:
        for w in list(self._watchers):
            w(event, self)

    def add_item(self, item: LineItem) -> "Order":
        if self.state is not OrderState.NEW:
            raise RuntimeError(f"{self.state.value} 状态的订单不能添加商品")
        self.items.append(item)
        self._notify("item.added")
        return self

    def pay(self) -> "Order":
        if self.state is not OrderState.NEW:
            raise RuntimeError("订单不可重复支付")
        self.state = OrderState.PAID
        self._notify("order.paid")
        return self

    def finish(self) -> "Order":
        self.state = OrderState.DONE
        self._notify("order.done")
        return self

    def __str__(self) -> str:
        return f"Order({self.no}, {self.state.value}, {self.total})"

# 5) 使用
def audit(event: str, order: Order) -> None:
    print(f"  [审计] {order.no}: {event} @ {datetime.now():%H:%M:%S}")

o = Order("NO-2024-001")
o.watch(audit)
o.add_item(LineItem("机械键盘", Decimal("299.00"), 1))
o.add_item(LineItem("鼠标垫", Decimal("49.90"), 2))
print(o)

o.rule = MemberPrice()          # 热切换定价策略
print("会员价:", o.total)

o.pay().finish()
print(o)
try:
    o.add_item(LineItem("鼠标", Decimal("99.00")))
except RuntimeError as e:
    print("拒绝:", e)

# 6) 换一条规则：包邮门槛
FREE_SHIPPING = Decimal("299")
print("是否包邮:", o.total >= FREE_SHIPPING)
print("创建于", o.created_at.strftime("%Y-%m-%d %H:%M:%S"))

输出（时间部分略有不同）：

  [审计] NO-2024-001: item.added @ 10:22:31
  [审计] NO-2024-001: item.added @ 10:22:31
Order(NO-2024-001, new, 397.80)
会员价: 358.02
  [审计] NO-2024-001: order.paid @ 10:22:31
  [审计] NO-2024-001: order.done @ 10:22:31
Order(NO-2024-001, done, 358.02)
拒绝: done 状态的订单不能添加商品
是否包邮: True
创建于 2026-10-18 10:22

设计模式选型速查与常见反模式

下面这张表是本章的浓缩版，建议打印贴在显示器旁：

12.1 反模式速查

1. 贫血对象（Anemic Object Model）：类只有字段，所有逻辑都在函数里。→ 用 dataclass + 领域方法把行为收回去。
2. 上帝类（God Object）：一个类几百个属性、上百个方法，什么都管。→ 按职责拆类，用组合连接。
3. 继承用来实现复用：B 继承 A 只是为了「用 A 的方法」，B 并不是一种 A。→ 改为持有 A 的实例。
4. 为一条 if 写一个模式：三个分支就上策略模式。→ 早期用 if，重复第三次再抽。
5. 反射式魔法：大量 __getattr__ / setattr 做动态字段。→ 会让 IDE、mypy、序列化全失效。
6. 可变默认参数与可变类属性：class C: items = [] 被所有实例共享。→ 用 field(default_factory=list)。

12.2 什么时候该停止抽象

模式不是越多越好。判断标准很简单：同样的代码结构在第三个地方出现之前，不要抽。反过来说，当你发现自己在写「万能基类」「万能配置对象」，几乎总是抽象过度了——拆回去，你会更快。

本章小结

• property 把方法伪装成属性，能在不破坏调用方的前提下加校验、加派生字段；functools.cached_property 给昂贵的计算加缓存，但要求实例有 __dict__（不能与 __slots__ 直接共存）。
• 混入类（Mixin） 无状态、靠 super() 协作；定义了 __init__ 就必须调用 super().__init__()。多继承的实际顺序由 C3 线性化算出的 MRO 决定，用 cls.mro() 查看。
• 组合优于继承：用组合 + Protocol 依赖注入，运行时可替换、测试可造假对象；继承层次超过 3 层就该考虑重构。
• 单例 在 Python 里有四种实现：模块级变量（最推荐）、__new__、装饰器、Enum。真正的问题是「你确定需要全局唯一吗」。
• 工厂模式 把实例化逻辑集中：简单工厂用字典查表，注册式工厂用装饰器自动登记，抽象工厂保证一族对象风格一致。
• 观察者模式 用于一对多通知；注意回调里修改订阅列表要遍历副本，弱引用订阅时绑定方法必须额外保活，订阅者抛异常要隔离以免拖垮全场。
• 策略模式 三个层级：传函数（最 Pythonic）、注册表 + 策略对象、Context 持有并可热切换。简单场景直接用 lambda 或 match。
• dataclass 自动生成 __init__/__repr__/__eq__；必记 field(default_factory=list)，以及 frozen/order/slots/kw_only 四个常用开关。
• Enum 用 auto() 简化、@unique 禁别名、IntFlag 做权限位组合；持久化场景务必显式写死值。
• 描述符 是 property 的可复用版，同一条校验规则出现在 5 个以上类里时就该抽出来。
• 贯穿全章的原则：组合优于继承、依赖注入优于硬编码、显式优于隐式、简单优于炫技。

练习题

1. 矩阵类（综合运算符重载）：实现 Matrix 类，支持 +、-、*（数乘与矩阵乘）、==、len()、repr，以及 __getitem__ 支持二维下标 m[i][j] 和 m[i, j] 两种写法。用 @dataclass 实现，并保证矩阵形状不匹配时抛出带清晰信息的 ValueError。

   参考答案要点内部用 list[list[float]] 存储；__add__ 用 zip；矩阵乘用三重循环或 zip(*b) 转置后点积；__getitem__ 判断参数是 tuple 还是 int；__eq__ 要先比形状。务必处理 other 类型不匹配时 return NotImplemented。

2. 受控属性：给 Rectangle 类加 property：width/height 写入时必须为正数，否则抛 ValueError；area 是只读计算属性；scaled(k) 返回缩放后的新矩形（原对象不变）。再实现 __repr__ 使其可被 eval 还原。

   参考答案要点@width.setter 内部赋给 self._width；area 只写 getter 就会自动只读；scaled 返回 Rectangle(self.width*k, self.height*k)。用 dataclass(frozen=True) + __post_init__ 校验也可以，但那样属性不可再赋值。

3. 日志混入类：实现 LoggableMixin，在类初始化时打印创建日志，并为每个被调用的公共方法自动打印「进入/退出」。要求通过 __init_subclass__ 或装饰器包裹方法实现，且子类没有 __init__ 参数污染。

   参考答案要点用类装饰器遍历 cls.__dict__，对以 _ 开头的非 __init__ 方法用 functools.wraps 包装并打印日志；__init__ 接受 *args, **kwargs 并调用 super().__init__(*args, **kwargs)；用 time.perf_counter() 记录耗时。

4. 配置管理器单例：实现一个 ConfigManager 单例，支持从 JSON 文件加载配置、get(key, default) 读取、set(key, value) 写入并即时落盘。要求：① 多线程下首次创建只执行一次；② 提供 reset() 便于单元测试。

   参考答案要点用 __new__ + threading.Lock 做双重检查；__init__ 用 hasattr(self, "_loaded") 防重复执行；写盘用 json.dump 到临时文件再 os.replace 原子替换；reset() 把类变量 _instance 置 None。也可以改用模块级单例，更简单。

5. 形状工厂 + 面积统计：用注册式工厂（@register("circle") 装饰器）支持 circle、rect、triangle 三种形状，每个都有 area() 和 perimeter()。然后实现 total_area(shapes)，要求传入「工厂名 + 参数」的字典列表并返回 Decimal 总面积，未知名字时抛 KeyError。

   参考答案要点用 @dataclass 定义各形状，register 内部查 _REGISTRY 并防止重名；面积用 Decimal(str(...)) 避免浮点误差；三角形形状不合法（边长不满足三角不等式）时抛 ValueError。

6. 观察者：状态机 + 日志：实现 Connection 类，状态在 DISCONNECTED → CONNECTING → CONNECTED → CLOSING → CLOSED 之间迁移。每次迁移调用 connect()/close() 时，通知所有注册的处理函数，并阻止非法迁移（如 CONNECTED 直接跳到 CONNECTING）。处理函数抛异常不能影响状态迁移。

   参考答案要点用 Enum 定义状态 + ALLOWED: dict[State, set[State]] 定义迁移表；subscribe(fn) 存列表；_notify 遍历 list(...) 副本并用 try/except 隔离每个回调的异常并打印警告。

7. 策略：压缩算法可切换：实现 Compressor 类，支持 gzip、zlib、none 三种策略（用 Protocol 声明接口），提供 compress(data: bytes) 与 decompress(blob)，支持运行时通过属性或方法切换策略。要求 none 策略原样返回，验证压缩后数据可无损还原。

   参考答案要点Protocol 里声明 compress/decompress 两个方法即可，无须继承；用 gzip.compress/gzip.decompress、zlib.compress/zlib.decompress；Context.switch(name) 内部查注册表；测试用 assert decompressor.decompress(blob) == original。注意 none 策略解压时也要能识别，最好给压缩数据加个 1 字节头部标记算法。

8. dataclass 全特性：实现 Product，要求：slots=True；password 或 token 类敏感字段不出现在 repr 中；id 使用 kw_only；创建时校验 price > 0 且 stock >= 0；支持 < 比较（按价格）；frozen 不需要但要能说明取舍。再实现 Warehouse（继承 Product）并解释字段顺序问题。

   参考答案要点@dataclass(slots=True)；敏感字段 field(repr=False, default="")；id: int = field(kw_only=True)；校验放 __post_init__；比较加 @dataclass(order=True) 或自定义 __lt__；继承时若父类字段都带默认值，子类新字段也必须带默认值，否则把父类字段改成 kw_only。