KEEL · 龙骨 · A CURRICULUM FOR THE AI ERA

02 · 模板方法与抽象基类:把「不变的流程」锁死在基类 — keel 龙骨

问题:CSV 导出和 JSON 导出是两种任务,但它们的「生命周期」一模一样——开文件、循环读记录、写每一行、统计、异常处理、收尾。把这套流程写两遍,将来改异常逻辑就得记得改两处。生产代码的解法是模板方法模式:骨架在基类,变化点留给子类。

问题:CSV 导出和 JSON 导出是两种任务,但它们的「生命周期」一模一样——开文件、循环读记录、写每一行、统计、异常处理、收尾。把这套流程写两遍,将来改异常逻辑就得记得改两处。生产代码的解法是模板方法模式:骨架在基类,变化点留给子类。


一、先看骨架:BaseExporter 与抽象方法

from abc import ABC, abstractmethod

class BaseExporter(ABC):                      # ← ABC:抽象基类,只用于继承
    """导出基类:管通用的生命周期(资源、循环、统计、异常、收尾)。"""

    def __init__(self, out_path: str):
        self.out_path = out_path

    @abstractmethod                            # ← 抽象方法:子类必须实现
    def _read_rows(self):
        """提供数据源,逐行 yield 出记录(dict)。"""
        ...

    @abstractmethod
    def _format_row(self, row: dict) -> str:
        """把一行格式化成字符串——CSV/JSON 各自不同。"""
        ...

三个语言点:

  1. ABC(Abstract Base Class):标记这是个「半成品类」,只用于继承;
  2. @abstractmethod:子类不实现 _read_rows / _format_row 就无法实例化——CsvExporter() 时直接 TypeError。这是把「子类必须提供这个钩子」从口头约定变成语言检查;
  3. 抽象方法就是基类的「留白」——它规定「你必须有这一手」,但不规定「你怎么干」。

二、骨架如何调用肉:export() 模板方法

class BaseExporter(ABC):
    # ... 上面的抽象方法 ...

    def export(self) -> int:                   # ← 模板方法:对子类不可见的手
        count = 0
        try:
            with open(self.out_path, "w", encoding="utf-8") as fh:
                for row in self._read_rows():  # ① 公共循环
                    line = self._format_row(row)   # ② ★ 钩子:格式由子类决定
                    fh.write(line)
                    count += 1                 # ③ 公共统计
        except Exception as exc:
            self._handle_error(exc)            # ④ 公共异常处理
            raise
        else:
            self._finalize(count)              # ⑤ 公共终态
        return count

    def _handle_error(self, exc: Exception) -> None:
        print(f"[export] failed: {exc}")

    def _finalize(self, count: int) -> None:
        print(f"[export] wrote {count} rows to {self.out_path}")

模式结构:export() 是那只「对子类不可见的手」——它规定「先开资源、循环里查什么、异常怎么办、最后怎么收尾」;唯一的「留白」是 ①② 处的钩子。子类填什么格式,骨架就跑什么。

三、子类怎么填空

class CsvExporter(BaseExporter):
    def _read_rows(self):
        yield {"id": 1, "name": "alice"}
        yield {"id": 2, "name": "bob"}

    def _format_row(self, row: dict) -> str:
        return f"{row['id']},{row['name']}\n"    # CSV 格式

class JsonExporter(BaseExporter):
    def _read_rows(self):
        yield {"id": 1, "name": "alice"}
        yield {"id": 2, "name": "bob"}

    def _format_row(self, row: dict) -> str:
        import json
        return json.dumps(row) + "\n"            # JSON 格式

两种导出共享 90% 的生命周期代码——这就是模板方法的账:多写一个抽象基类,换来「改异常逻辑只改一处」「新增一种导出格式只填一个空」。

四、模板方法 vs 其他组合方式(选型)

方案 什么时候用 缺点
模板方法(本例) 流程骨架固定、多个「变化点」集中在几步 继承是静态的,运行时换不了骨架
策略模式(传函数/对象) 变化点要在运行时切换 骨架本身也要跟着变时不适用
装饰器包装 只想增强「进出」两侧,不改中间流程 中间循环里插逻辑(如统计)不方便
复制粘贴 ❌ 上面已经说过了

这个例子选模板方法的理由就藏在骨架里:开文件、统计、异常、收尾都是插在循环外面/里面的公共逻辑,装饰器够不着;而「格式怎么写」在构造时已确定,不需要运行时切换。

4.1 一个细节:为什么把「产生数据」和「消费数据」拆开?

如果子类直接自己 for row in source: fh.write(...) 把循环写完,异常处理/统计/收尾就得在每个子类里重复——骨架就退化成了「建议」。把「提供数据源」(_read_rows)和「消费数据」(基类 export 的循环)拆开,才锁死了公共逻辑的唯一性。这个「生产与消费分离」的思想在事件管道类系统里反复出现。

五、可迁移的套路

  1. 多条业务线共享同一个生命周期(开始→循环→终态→收尾)→ 抽象基类 + 模板方法,@abstractmethod 锁死扩展点;
  2. 公共逻辑插在循环中间时,别用装饰器/中间件硬绕,让子类只提供「数据源」或「每步钩子」;
  3. 子类构造函数先 super().__init__(公共字段),再加自己的——字段归属清晰;
  4. 判断一个项目里模板方法用得好不好:新增一种业务类型时,要不要碰基类? 好的设计的答案是不要,只写新子类。

↓下一步:03 章 · 闭包与延迟导入

进入 keel 阅读