Python · 类、MRO 与对象模型

场景 1 · dataclass 定义订单 DTO: frozen 防篡改 + default_factory 防共享

DTO 在服务边界被几十个函数传递, 谁都能改一笔就是隐患。frozen 让写入当场炸, 可变默认值交给工厂:

from dataclasses import dataclass, field

@dataclass(frozen=True, slots=True)           # frozen: 下游偷改订单直接炸
class OrderDTO:
    order_id: str
    items: tuple[Item, ...]                     # 不可变集合配 frozen
    tags: frozenset = frozenset()                   # 不可变类型才配当默认值
    meta: dict = field(default_factory=dict)   # 每实例独立 dict

# o.items = (...) → FrozenInstanceError — 配置错误当场暴露, 不带病上线

场景 2 · 千万实例网格: __slots__ 内存减半的量化案例

风控回测要同时持有 1000 万个行情点对象, 默认 __dict__ 版直接把 512MB 容器打爆:

class PointDict:                       # 默认: 每实例背一个 __dict__
    def __init__(self, x, y): self.x, self.y = x, y

class PointSlots:
    __slots__ = ("x", "y")              # 定长槽, 没有 __dict__
    def __init__(self, x, y): self.x, self.y = x, y

# tracemalloc 实测 10_000_000 实例:
#   PointDict  ≈ 1.62 GB  →  PointSlots ≈ 0.76 GB   (-53%)
#   属性访问快 ~10%; 代价: 不能动态加属性 (海量小对象求之不得)

海量小对象 (网格/词表/回测 bar) 一律 slots; 数量级到不了百万就别为它牺牲灵活性。

场景 3 · 单例/连接持有者的三种实现与取舍

全局只要一个的重量级对象, 三种写法各有成本 — 生产默认选最朴素的:

# ① 模块级 (Python 之禅): 模块只执行一次, config = load() 天然单例
#    优点零魔法; 缺点 import 即初始化, 测试要多实例得 hack

# ② __new__ 拦截: 需要"延迟到首次使用"时
class ConnHub:
    _inst = None
    def __new__(cls):
        if cls._inst is None:              # 首次才真正创建
            cls._inst = super().__new__(cls)
        return cls._inst

# ③ metaclass: 所有子类自动单例 — 框架级魔法, 业务别碰
# 生产默认①; 要多环境/易测试 → 工厂 + 显式依赖注入

场景 4 · 自定义容器: 两个魔术方法白得全套序列能力

播放列表类想被 len/切片/for/in 像原生列表一样对待 — 不用继承 list, 实现协议即可:

class Playlist:
    def __init__(self, *tracks): self._t = list(tracks)
    def __len__(self): return len(self._t)
    def __getitem__(self, i):
        if isinstance(i, slice):             # 支持 pl[10:20] 切片
            return Playlist(*self._t[i])
        return self._t[i]
    def __iter__(self): return iter(self._t)

# 白得的能力: len(pl), pl[3], pl[1:5], for t in pl, "x" in pl, list(pl)

场景 5 · Mixin 组合能力 vs 多重继承翻车的重构

三个业务类都要 JSON 导出与审计日志。老代码从公共基类继承, 基类越长越不敢动; 重构成窄 Mixin:

class JSONMixin:                          # 能力: 可序列化; 无状态, 不单用
    def to_json(self):
        return json.dumps(self.__dict__, default=str, ensure_ascii=False)

class AuditMixin:
    def audit(self, actor):                    # 只依赖宿主自带属性
        log.info("audit", obj=type(self).__name__,
                 oid=getattr(self, "id", None), actor=actor)

class Report(JSONMixin, AuditMixin):        # 声明能力而非继承实现
    ...

# Mixin 排 MRO 最左; 动了能力不影响业务类, 各能力可独立测试

场景 6 · ABC 强制子类实现接口: 报错提前到启动期

存储层要换 Redis/MySQL 双实现, 新人漏实现一个方法要到运行到那行才炸。抽象基类把契约钉死:

from abc import ABC, abstractmethod

class Storage(ABC):
    @abstractmethod
    def get(self, key): ...
    @abstractmethod
    def put(self, key, val): ...

class RedisStorage(Storage):
    def get(self, key): return self.r.get(key)
    # 忘了 put → RedisStorage() 当场 TypeError:
    # Can't instantiate abstract class without put — 启动期报错

场景 7 · __enter__/__exit__ 让对象支持 with 计时

给每个 ETL 阶段统一计时打点, 不想在每个函数里复制 try/finally — 做成上下文管理器对象 (机制详见上下文主题):

class timer:                                    # 用法: with timer("etl"): ...
    def __init__(self, name): self.name = name
    def __enter__(self):
        self.t0 = time.perf_counter(); return self
    def __exit__(self, exc_t, exc_v, tb):
        ms = (time.perf_counter() - self.t0) * 1000
        metrics.histogram("phase.ms", ms, phase=self.name)  # 异常也计时
        return False                                # 不吞异常

场景 8 · property 做字段校验与惰性计算

余额字段要"负数进不来 + 首次访问才查账本"。门面用 property, 内部走私有属性, 一道门收口:

class Account:
    def __init__(self, aid): self.aid, self._balance = aid, None

    @property
    def balance(self):                          # 惰性: 首次访问才查账本
        if self._balance is None:
            self._balance = self._fetch_from_ledger()
        return self._balance

    @balance.setter
    def balance(self, v):
        if v < 0: raise ValueError("balance cannot be negative")
        self._balance = v                       # 写内部字段, 不触发自身

场景 9 · __repr__ 让重试日志一眼可读

重试框架打日志全是 "object at 0x7f3a...", 排障时对不上是哪个任务。手写 repr 是性价比最高的投资:

@dataclass
class Task:
    id: str; status: str; attempt: int = 0
    def __repr__(self):                        # 排障字段全带上, 截断防刷屏
        return f"Task(id={self.id!r}, status={self.status}, try={self.attempt})"

# 日志: [retry] Task(id='t-91', status=failed, try=3) — 一眼定位, 可直接 grep
# 默认的 <Task object at 0x7f...> 在排障现场等于什么都没打

场景 10 · __setstate__ 兼容旧版本存档的字段演进

缓存/存档里的对象是旧结构, 新代码加了字段, 反序列化就缺 key。用状态钩子做迁移 (序列化细节见序列化主题):

class Profile:
    def __setstate__(self, state):
        if "created_at" not in state:         # 旧版本存档没有该字段
            state["created_at"] = utcnow()          # 迁移默认值
        if "tags" not in state:
            state["tags"] = []
        self.__dict__.update(state)              # 永远兼容旧快照

# dataclass 等价做法: 新字段给 default; __init__ 缺字段时不会炸