pythonABC抽象模块

Python 抽象基类 ABC 与 @abstractmethod

适用场景：你正在写一个会被多人协作、长期演进的 Python 项目，需要让”一组类必须长得一样”。 阅读路径：第一部分建立直觉，第二部分掌握语法，第三部分拆底层机制，第四部分是可直接落地的企业级案例。 文中所有代码均在 Python 3.13.15 实测通过，关键输出直接附在代码之后。

第一部分 入门：为什么需要抽象基类

1.1 鸭子类型：自由度与它的账单

Python 的默认哲学是鸭子类型——不关心对象是什么类，只关心它有没有你要的方法。这让早期开发很快，代价是约定只存在于人脑和文档里，一旦有人记错，问题往往拖到运行期才爆出来。

先看一个再普通不过的报表模块：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class CSVDataSource:

    def fetch_data(self):
        return "csv-data"


class APIDataSource:

    def get_data(self):  # 注意：名字和上面不一样
        return "api-data"


def load(source):
    return source.fetch_data()  # 这里只认 fetch_data

调用时：

1
2
3
print(load(CSVDataSource()))  # csv-data
print(load(APIDataSource()))

第二行会在运行时抛出：

1
2
AttributeError: 'APIDataSource' object has no attribute 'fetch_data'

这里暴露了三个问题。命名没有统一约束，fetch_data 和 get_data 谁对谁错全凭记忆；错误暴露得太晚，要等到真去调用才发现；新增数据源时没有任何”照着填”的模板，漏掉哪个方法只能靠运气。项目小的时候这些问题还能忍，类一多、人一多就开始互相拆台。

1.2 用 NotImplementedError 打补丁，以及它的漏洞

一个流传很广的做法，是在基类里把方法体写成抛异常：

1
2
3
4
5
6
7
8
9
class BaseProcessor:

    def process(self, data):
        raise NotImplementedError("子类必须实现 process()")


class Broken(BaseProcessor):
    pass

实测行为是：

1
2
3
b = Broken()  # 实例化成功，什么都不报
b.process("x")  # 直到这一行才爆炸

1
2
3
实例化成功（埋雷）: <Broken object at 0x...>
调用时才爆炸: 子类必须实现 process()

它能跑，但约束是”软”的。漏实现的对象照样能被创建、被传递、被存进数据库，直到某条业务链路真正调用到它才崩。线上事故常常就是这么来的——process 可能只在某个分支里被调用，测试没覆盖到，问题就一路溜到生产。

1.3 ABC + @abstractmethod：把契约提前到实例化那一刻

标准库 abc 模块提供了官方方案：抽象基类（Abstract Base Class）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from abc import ABC, abstractmethod


class BaseProcessor(ABC):

    @abstractmethod
    def process(self, data):
        ...


class Broken(BaseProcessor):
    pass


b = Broken()

这一行会直接失败：

1
2
3
TypeError: Can't instantiate abstract class Broken without an implementation
for abstract method 'process'

区别就在于报错的时机。NotImplementedError 要等调用，ABC 在创建对象时就拦下了。对象都造不出来，自然没机会流进业务逻辑。

1.4 逐行拆解第一个完整的抽象基类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from abc import ABC, abstractmethod
import math


class Shape(ABC):  # (1)

    @abstractmethod  # (2)
    def area(self) -> float:  # (3)
        ...

    @abstractmethod
    def perimeter(self) -> float:
        ...

    def describe(self):  # (4)
        return f"面积={self.area():.2f} 周长={self.perimeter():.2f}"


class Circle(Shape):  # (5)

    def __init__(self, r):
        self.r = r

    def area(self):
        return math.pi * self.r**2

    def perimeter(self):
        return 2 * math.pi * self.r

逐点说明：

  1. Shape 继承 ABC。这个继承的意义是让 Shape 的元类变成 ABCMeta，从而启用”抽象方法检查”这套机制。它不是功能增强，而是告诉解释器”这个类是契约，不是成品”。
  2. @abstractmethod 给方法打上”必须被实现”的标记。它只做标记，不改变函数本身的行为。
  3. 抽象方法体写 ... 或 pass 都行。Python 不禁止你写真实逻辑，后续 2.3 会讲这有什么用。
  4. describe 是普通的具体方法，带有默认实现。子类直接继承，不用重写。抽象基类可以同时容纳”必须实现”和”已经写好”的部分。
  5. Circle 实现了两个抽象方法，于是它成为了”具体类”，可以实例化。

运行结果：

1
2
3
4
print(Circle(5).describe())  # 面积=78.54 周长=31.42

Shape()  # TypeError: 抽象类不能直接实例化

Shape() 抛出：

1
2
3
TypeError: Can't instantiate abstract class Shape without an implementation
for abstract methods 'area', 'perimeter'

如果子类只实现了一半：

1
2
3
4
5
class Incomplete(Shape):

    def area(self):
        return 0.0

实例化时报错会精确点出缺的是哪个：

1
2
3
TypeError: Can't instantiate abstract class Incomplete without an implementation
for abstract method 'perimeter'

1.5 怎么读这些报错

报错信息里有两个关键信息点。一是缺哪个方法，abstract method 'perimeter' 直接告诉你少写了谁；二是发生在实例化时，说明这不是写错了逻辑，而是根本没满足契约。以后看到 Can't instantiate abstract class，先去看子类是不是漏实现了某个抽象成员，八成就在那里。

第二部分 进阶：ABC 的六大能力

2.1 两种定义方式：继承 ABC 还是指定元类

1
2
3
4
5
6
7
8
9
10
from abc import ABC, ABCMeta


class Way1(ABC):  # 推荐
    pass


class Way2(metaclass=ABCMeta):  # 等价写法
    pass

两者完全等价。abc.ABC 本身只是一个空壳类，它的元类被设成了 ABCMeta：

1
2
type(Way1).__name__  # 'ABCMeta'

推荐第一种写法。第二种在你已经有别的元类、需要手工组合时才会用到（见 3.1），平时没必要引入 metaclass= 这种更容易出错的语法。

2.2 @abstractmethod 到底做了什么

@abstractmethod 的实现非常轻。它给函数对象挂一个属性：

1
2
3
4
def abstractmethod(funcobj):
    funcobj.__isabstractmethod__ = True
    return funcobj

真正的检查发生在类创建阶段。ABCMeta 在生成类时，会扫描命名空间和所有基类，把所有带 __isabstractmethod__ = True、且当前类没有覆盖的名字，收集进一个集合 __abstractmethods__。名字还在集合里，类就算抽象类，实例化时被拦下。

动手看一眼这个集合：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class ABC1(ABC):

    @abstractmethod
    def a(self):
        ...

    @abstractmethod
    def b(self):
        ...


class Partial(ABC1):

    def a(self):
        return 1
    # b 没实现

1
2
print(Partial.__abstractmethods__)  # frozenset({'b'})

__abstractmethods__ 是一个 frozenset。它非空，就说明还有没兑现的契约。这也是各框架判断”类是否抽象”的底层依据——inspect.isabstract(cls) 内部就是看这个集合。

2.3 抽象方法可以有方法体，也能被 super() 调用

很多人以为抽象方法必须空着，其实不是。抽象方法完全可以写实现，子类通过 super() 拿到它：

1
2
3
4
5
6
7
8
9
10
11
12
class Base(ABC):

    @abstractmethod
    def step(self):
        return "base-step"


class Impl(Base):

    def step(self):
        return "impl+" + super().step()

1
2
Impl().step()  # 'impl+base-step'

这个特性在真实项目里很有用：把”每个子类都该做的公共部分”放进抽象方法体，子类先调用 super() 再补充自己的逻辑，既强制了实现，又复用了代码。

2.4 抽象属性

抽象成员不限于方法，属性也能抽象。装饰器的叠加顺序同样是 @abstractmethod 在最内层：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class Repository(ABC):

    @property
    @abstractmethod
    def table_name(self) -> str:
        ...


class UserRepo(Repository):

    @property
    def table_name(self):
        return "users"

1
2
3
print(UserRepo().table_name)  # 'users'
Repository()  # TypeError

子类必须提供一个同名的 property，光实现成普通方法也不行——类型对不上。

2.5 抽象类方法与抽象静态方法（顺序写反会直接报错）

@classmethod、@staticmethod 可以和 @abstractmethod 叠加，但顺序有硬性要求：@abstractmethod 必须写在最靠近 def 的位置（最内层）。

正确写法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class Codec(ABC):

    @classmethod
    @abstractmethod
    def name(cls) -> str:
        ...

    @staticmethod
    @abstractmethod
    def encode(data: str) -> bytes:
        ...


class UTF8Codec(Codec):

    @classmethod
    def name(cls):
        return "utf-8"

    @staticmethod
    def encode(data):
        return data.encode("utf-8")

1
2
3
print(UTF8Codec.name())  # utf-8
print(UTF8Codec.encode("中"))  # b'\xe4\xb8\xad'

如果顺序写反：

1
2
3
4
5
6
7
class Wrong(ABC):

    @abstractmethod
    @classmethod
    def name(cls):
        ...

在 Python 3.13 下，这个类定义的时候就报错：

1
2
AttributeError: attribute '__isabstractmethod__' of 'classmethod' objects is not writable

静态方法反序同理（'staticmethod' objects is not writable）。原因是装饰器自下而上执行：@classmethod 先把函数包成 classmethod 对象，@abstractmethod 再去给它设 __isabstractmethod__，而这个属性不可写。

顺带说一句历史包袱。abc 早期提供过 abstractproperty、abstractclassmethod、abstractstaticmethod 三个独立装饰器。它们从 Python 3.3 起就已弃用，现在还在标准库里但不要再用了，统一改成上文的叠加写法即可。

2.6 模板方法模式：具体方法 + 抽象步骤

这是 ABC 最有价值的一种用法，也是设计模式里的”模板方法”。把固定流程写在具体方法里，把可变步骤定义成抽象方法，交给子类填。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class DataPipeline(ABC):

    def run(self, data):  # 固定骨架
        cleaned = self.clean(data)
        result = self.transform(cleaned)
        self.save(result)
        return result

    @abstractmethod
    def clean(self, data):
        return data

    @abstractmethod
    def transform(self, data):
        ...

    @abstractmethod
    def save(self, data):
        ...

子类只需要关心自己的那几步：

1
2
3
4
5
6
7
8
9
10
11
12
class UpperPipeline(DataPipeline):

    def clean(self, data):
        return data.strip()

    def transform(self, data):
        base = super().transform(data)
        return data.upper() if base is None else base

    def save(self, data):
        print("已保存:", data)

1
2
UpperPipeline().run("  hello  ")

输出：

1
2
3
已保存: HELLO
结果: HELLO

run 里的编排逻辑只写一次，所有子类共享同一条流水线。以后要在流程里插入埋点或审计，只改基类一处，不用挨个改子类。

2.7 isinstance / issubclass 与虚拟子类 register()

抽象基类天然参与类型判断：

1
2
3
issubclass(Circle, Shape)  # True
isinstance(Circle(5), Shape)  # True

更灵活的是”虚拟子类”。假设有个现成的旧类没继承你的基类，你又不想改它的源码：

1
2
3
4
5
6
7
8
9
10
11
12
class Notifier(ABC):

    @abstractmethod
    def send(self, msg):
        ...


class LegacySMS:  # 没继承 Notifier

    def send(self, msg):
        print("legacy sms:", msg)

1
2
3
4
5
print(issubclass(LegacySMS, Notifier))  # False
Notifier.register(LegacySMS)
print(issubclass(LegacySMS, Notifier))  # True
print(isinstance(LegacySMS(), Notifier))  # True

注册之后，isinstance 和 issubclass 都会认账。有两点要注意：

1
2
print(Notifier in LegacySMS.__mro__)  # False

虚拟子类不会进入 MRO，也拿不到基类的具体方法，它只是让类型判断通过。另外 register 不做任何方法检查——上面 LegacySMS 恰好有 send，就算没有，注册照样成功。所以它是”声明式”的，责任在注册的人身上。

2.8 subclasshook：让类型判断认”结构”不认”出身”

如果想让判定更聪明——不看你继承没继承，只看你有没有某个方法，可以重写 __subclasshook__。这正是标准库 collections.abc 的做法，也是 issubclass(tuple, abc.Sequence) 能返回 True 的原因：tuple 是 C 实现，根本没机会去继承 Python 的 Sequence。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class MyIterable(ABC):

    @classmethod
    def __subclasshook__(cls, C):
        if cls is MyIterable:
            if any("__iter__" in B.__dict__ for B in C.__mro__):
                return True
        return NotImplemented


class MyList:

    def __iter__(self):
        return iter([])

1
2
3
4
print(issubclass(MyList, MyIterable))  # True
print(issubclass(list, MyIterable))  # True
print(issubclass(tuple, MyIterable))  # True

返回值有三种含义：True 表示认定为子类，False 表示认定为不是（即使它本来就继承了你），NotImplemented 表示交回默认机制继续判断。写这个钩子时务必定义成 classmethod，否则行为不稳定。

第三部分 精通：底层机制与选型

3.1 ABCMeta 的类创建流程

理解 ABCMeta 的时间线，能解释后面所有的”坑”。它继承自内置的 type，核心动作在 __new__：

1
2
3
4
5
6
7
class ABCMeta(type):

    def __new__(mcls, name, bases, namespace, **kwargs):
        cls = super().__new__(mcls, name, bases, namespace, **kwargs)
        _abc_init(cls)  # 在这里算出 __abstractmethods__、处理寄存器
        return cls

顺序很关键：super().__new__(...) 先执行，它会触发 __init_subclass__；_abc_init 在后面才算出 __abstractmethods__。记住这条时间线，下一个坑就顺理成章了。

另外，当你同时需要别的元类（比如 enum、ORM 的元类）时，得手工做一个同时继承 ABCMeta 的元类，因为 Python 只允许一个”最派生”的元类参与实例化。

3.2 一个几乎人人会踩的坑：abstractmethods 是 type 的描述符

先看现象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class Base(ABC):

    def __init_subclass__(self, **kwargs):
        super().__init_subclass__(**kwargs)
        print(getattr(cls, "__abstractmethods__", None))

    @abstractmethod
    def work(self):
        ...


class StillAbstract(Base):
    pass

输出是：

1
2
3
[__init_subclass__] StillAbstract 读到 __abstractmethods__ = None
类创建完成后再看: frozenset({'work'})

在 __init_subclass__ 里完全读不到抽象方法集合。原因有两层。第一层是时间线：__init_subclass__ 触发时 _abc_init 还没跑，集合还没算出来。第二层更隐蔽：

1
2
print("__abstractmethods__" in type.__dict__)  # True

__abstractmethods__ 是定义在 type 上的一个描述符，它只读当前类自己那份值，不会沿着父类 MRO 去找。所以别说读不到父类的，连”继承下来”这个概念都不成立——它是每个类各自独立的。

结论：任何依赖 __abstractmethods__ 的逻辑，都必须等类创建完成之后再执行。__init_subclass__ 不是合适的时机，得用自定义元类的 __init__（它跑在 _abc_init 之后）。第四部分的插件案例会用上这一点。

3.3 分层抽象与多重继承

抽象类可以继续派生抽象类，逐层细化契约：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class L1(ABC):

    @abstractmethod
    def a(self):
        ...

    @abstractmethod
    def b(self):
        ...


class L2(L1):

    def a(self):
        return 1

1
2
3
print(L2.__abstractmethods__)  # frozenset({'b'})
L2()  # TypeError：还差 b

再补上 b，L3 就能实例化。这套机制让你可以先定义完整契约，再用中间层实现其中一部分，逐步逼近可用。

多重继承时，如果两个基类声明了同名抽象方法，子类实现一次就同时满足两者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class Readable(ABC):

    @abstractmethod
    def open(self):
        ...


class Writable(ABC):

    @abstractmethod
    def open(self):
        ...


class ReadWrite(Readable, Writable):

    def open(self):
        return "opened"

1
2
ReadWrite().__abstractmethods__  # frozenset()，两个契约一起满足

抽象基类也很适合当 mixin：提供若干具体方法，混进业务类里复用。

3.4 ABC、Protocol、鸭子类型：到底怎么选

Python 里描述”接口”有三条路，各有边界。

抽象基类（ABC）：靠显式继承建立契约，运行时会强制。子类没实现全，实例化就失败。适合你要控制继承体系、需要运行时拦截的场景：框架、插件、支付/存储这类适配层。

Protocol（typing，Python 3.8+）：靠结构判断，只在静态检查阶段生效，运行时不做任何拦截。对象只要有匹配的方法签名就算符合协议，哪怕它压根不知道协议的存在。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from typing import Protocol, runtime_checkable


@runtime_checkable
class Drawable(Protocol):

    def draw(self) -> str:
        ...


class Circle:  # 不继承 Drawable

    def draw(self):
        return "画个圆"

1
2
3
isinstance(Circle(), Drawable)  # True（加了 runtime_checkable）
issubclass(Circle, Drawable)  # True

注意 runtime_checkable 只检查方法名在不在，不校验签名。默认的 Protocol 连 isinstance 都会报错。

纯鸭子类型：什么约束都不加，最灵活也最容易失控。脚本、一次性工具用它没问题。

选型的一个实用判据：需要”运行时把不合格的对象挡在门外”就选 ABC；只需要”让类型检查器帮我抓错、又不想强迫别人继承”就选 Protocol；内部小工具、改动范围可控，就继续用鸭子类型。

3.5 update_abstractmethods 与 get_cache_token

update_abstractmethods(cls)（Python 3.10+）用于在类创建之后重新计算抽象状态。典型场景是动态给类挂方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
from abc import ABC, abstractmethod, update_abstractmethods


class Base(ABC):

    @abstractmethod
    def go(self):
        ...


class Dyn(Base):
    pass

1
2
3
4
5
6
print(Dyn.__abstractmethods__)  # frozenset({'go'})
Dyn.go = lambda self: "dynamic"
update_abstractmethods(Dyn)
print(Dyn.__abstractmethods__)  # frozenset()
print(Dyn().go())  # dynamic

get_cache_token() 返回一个令牌，每次调用 register() 都会让它变化。写缓存逻辑时可以用它判断”ABC 的注册表有没有变过”：

1
2
3
4
5
6
7
from abc import get_cache_token

t1 = get_cache_token()
Notifier.register(SomeClass)
t2 = get_cache_token()
assert t1 != t2

3.6 常见陷阱清单

  • 装饰器顺序写反：@abstractmethod 必须在最内层，否则定义时就报 AttributeError。
  • 以为抽象方法不能有实现：可以有，而且能被 super() 调用，是个好用的复用点。
  • 在 __init_subclass__ 里判断 __abstractmethods__：读不到，用自定义元类的 __init__。
  • 对抽象类用 super().__init__() 传参不当：抽象基类若有 __init__，子类记得正确调用。
  • 忘了 register 只影响类型判断：它不检查方法、不进 MRO、不给方法。
  • 抽象面开太大：抽象方法越多，子类负担越重，改契约越像改公共 API。

第四部分 企业级实战

以下五个案例都来自真实系统的常见形态，代码可直接运行。

案例 1 支付网关：策略 + 模板方法 + 异常契约

背景：系统要接入支付宝、微信、银联等多条渠道。每个渠道的初始化参数、请求方式都不同，但对外必须统一提供”预授权、扣款、退款”。同时，重试、异常映射这类编排逻辑不该让每个渠道各写一遍。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class PaymentError(Exception):
    """支付领域异常基类"""


class RetryableError(PaymentError):
    """可重试的基础设施错误（超时、限流等）"""


class PaymentGateway(ABC):
    """支付渠道统一契约"""

    timeout_seconds: float = 5.0  # 类属性，子类可覆盖

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

    @abstractmethod
    def authorize(self, order_id: str, amount: float) -> str:
        """预授权，返回授权流水号"""

    @abstractmethod
    def capture(self, auth_id: str, amount: float) -> dict:
        """扣款"""

    @abstractmethod
    def refund(self, order_id: str, amount: float) -> dict:
        """退款"""

    def charge(self, order_id: str, amount: float, retries: int = 2) -> dict:
        """模板方法：预授权 -> 扣款，统一重试与异常收敛"""
        last_err = None
        for attempt in range(1, retries + 1):
            try:
                auth_id = self.authorize(order_id, amount)
                return self.capture(auth_id, amount)
            except RetryableError as e:
                last_err = e
                print(f"  第 {attempt} 次重试...")
        raise PaymentError(f"扣款失败: {last_err}")

具体渠道只填三个原语：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class AlipayGateway(PaymentGateway):

    def authorize(self, order_id, amount):
        return f"ali-auth-{order_id}"

    def capture(self, auth_id, amount):
        return {
            "channel": "alipay",
            "auth": auth_id,
            "amount": amount,
            "status": "success",
        }

    def refund(self, order_id, amount):
        return {"channel": "alipay", "order": order_id, "refund": amount}

运行：

1
2
print(AlipayGateway("M001").charge("O1001", 99.9))

1
2
{'channel': 'alipay', 'auth': 'ali-auth-O1001', 'amount': 99.9, 'status': 'success'}

再用一个会抖动的渠道验证重试：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class FlakyGateway(PaymentGateway):
    _counter = 0

    def authorize(self, order_id, amount):
        FlakyGateway._counter += 1
        if FlakyGateway._counter < 3:
            raise RetryableError("网关超时")
        return "flaky-auth"

    def capture(self, auth_id, amount):
        return {"channel": "flaky", "amount": amount, "status": "success"}

    def refund(self, order_id, amount):
        return {"status": "refunded"}


print(FlakyGateway("M002").charge("O1002", 50.0, retries=3))

1
2
3
4
  第 1 次重试...
  第 2 次重试...
{'channel': 'flaky', 'amount': 50.0, 'status': 'success'}

这个案例体现了几件事。三个 @abstractmethod 把渠道能力钉死成契约，谁少实现一个就实例化不了；charge 是模板方法，重试和异常收敛写一遍就够；异常也做了契约化——RetryableError 和不可重试的 PaymentError 分开，调用方知道哪些能重试。这是生产系统里最容易被忽略的部分：一个方法名清楚但失败语义含糊的接口，不算真正的接口。

案例 2 ETL 数据管道：模板方法 + 组合

背景：数据团队要把不同来源的数据清洗、转换、写入不同目标。流程固定，两端和中间变换可变。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class DataSource(ABC):

    @property
    @abstractmethod
    def name(self) -> str:
        """数据源名称（抽象属性）"""

    @abstractmethod
    def read(self) -> list:
        """读取原始记录"""


class DataSink(ABC):

    @abstractmethod
    def write(self, records: list) -> int:
        """写入记录，返回条数"""


class ETLJob(ABC):

    def __init__(self, source: DataSource, sink: DataSink):
        self.source = source
        self.sink = sink

    def run(self):  # 固定骨架
        print(f"[ETL] 源={self.source.name}")
        raw = self.source.read()
        cleaned = self.clean(raw)
        transformed = self.transform(cleaned)
        n = self.sink.write(transformed)
        print(f"[ETL] 写入 {n} 条")
        return n

    def clean(self, records: list) -> list:
        """可选钩子：给了默认实现，子类可按需覆盖"""
        return [r for r in records if r]

    @abstractmethod
    def transform(self, records: list) -> list:
        """必填：业务转换"""

端到端跑一遍：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class CsvSource(DataSource):

    @property
    def name(self):
        return "csv://orders.csv"

    def read(self):
        return [" alice ", None, " bob "]


class MemorySink(DataSink):

    def __init__(self):
        self.data = []

    def write(self, records):
        self.data.extend(records)
        return len(records)


class UpperJob(ETLJob):

    def transform(self, records):
        return [r.strip().upper() for r in records]


sink = MemorySink()
UpperJob(CsvSource(), sink).run()
print("sink 内容:", sink.data)

1
2
3
4
[ETL] 源=csv://orders.csv
[ETL] 写入 2 条
sink 内容: ['ALICE', 'BOB']

这里用了三种不同性质的成员：transform 是必须实现的抽象方法；name 是必须实现的抽象属性；clean 是带默认实现的可选钩子。这种”必需的少、可选的给默认值”的分配方式，是成熟抽象基类的典型特征——契约要小而稳，扩展点要留够。

案例 3 插件系统：自定义元类自动注册

背景：主程序要支持第三方插件。理想效果是——开发者写一个插件类，框架自动发现并注册，不用手工维护一张映射表。

直觉写法是在 __init_subclass__ 里注册：

1
2
3
4
5
6
7
8
class Plugin(ABC):
    registry = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        if not getattr(cls, "__abstractmethods__", None):
            Plugin.registry[cls.plugin_name] = cls

但这个写法有问题。前面 3.2 讲过，__init_subclass__ 触发时 __abstractmethods__ 还没写入，getattr 拿到的是 None，于是连还没填满的抽象中间层也会被注册。实测会看到：

1
2
3
4
已注册插件: json
已注册插件: reverse
已注册插件: abstract_middle      # 错误的注册

正确做法是自定义一个继承 ABCMeta 的元类，把注册逻辑放到 __init__——它跑在 _abc_init 之后，此时集合已经算好。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class PluginMeta(ABCMeta):

    def __init__(cls, name, bases, namespace, **kwargs):
        super().__init__(name, bases, namespace, **kwargs)
        plugin_name = namespace.get("plugin_name")
        if plugin_name and not cls.__abstractmethods__:  # 只注册具体类
            Plugin.registry[plugin_name] = cls
            print(f"  已注册插件: {plugin_name}")


class Plugin(ABC, metaclass=PluginMeta):
    registry: dict = {}
    plugin_name: str = ""

    @abstractmethod
    def execute(self, payload: dict) -> dict:
        """执行插件逻辑"""

开发者按模板写插件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class JsonPlugin(Plugin):
    plugin_name = "json"

    def execute(self, payload):
        import json

        return {"ok": True, "data": json.dumps(payload)}


class ReversePlugin(Plugin):
    plugin_name = "reverse"

    def execute(self, payload):
        return {"ok": True, "data": payload.get("text", "")[::-1]}


class AbstractMiddleLayer(Plugin):
    plugin_name = "abstract_middle"  # 仍含未实现抽象方法 -> 不注册

    def extra(self):
        return 1

实测结果：

1
2
3
4
5
已注册插件: json
已注册插件: reverse
注册表: ['json', 'reverse']
{'ok': True, 'data': 'olleh'}

AbstractMiddleLayer 没被误注册，注册表干净。这个案例把”抽象方法的强制力”和”元类介入类创建过程”结合了起来，是 ABC 用得比较深的一种形态。

案例 4 仓储模式：泛型抽象基类

背景：数据访问层要在内存实现和数据库实现之间切换，业务代码只依赖接口。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from dataclasses import dataclass
from typing import Generic, TypeVar

T = TypeVar("T")


@dataclass
class User:
    id: int
    name: str


class Repository(ABC, Generic[T]):

    @abstractmethod
    def add(self, entity: T) -> T:
        ...

    @abstractmethod
    def get(self, entity_id: int) -> "T | None":
        ...

    @abstractmethod
    def delete(self, entity_id: int) -> None:
        ...

    def exists(self, entity_id: int) -> bool:
        """具体方法：所有仓储共享，不用各写一遍"""
        return self.get(entity_id) is not None

内存实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class InMemoryUserRepository(Repository[User]):

    def __init__(self):
        self._store: dict = {}

    def add(self, entity: User) -> User:
        self._store[entity.id] = entity
        return entity

    def get(self, entity_id: int):
        return self._store.get(entity_id)

    def delete(self, entity_id: int) -> None:
        self._store.pop(entity_id, None)

1
2
3
4
5
6
7
repo = InMemoryUserRepository()
repo.add(User(1, "张三"))
print(repo.get(1))  # User(id=1, name='张三')
print(repo.exists(1))  # True
repo.delete(1)
print(repo.exists(1))  # False

ABC 和 Generic 可以一起继承。抽象方法定义读写删三个原语，exists 作为基于 get 的具体方法放在基类，所有实现共享。这种”薄接口 + 基类复用”的搭配，是数据访问层的标准形态。

案例 5 通知系统：抽象工厂，呼应开篇

回到本文开头那段通知代码，用抽象基类把它收拾干净后是这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class Notifier(ABC):

    @abstractmethod
    def send(self, message: str) -> bool:
        ...


class EmailNotifier(Notifier):

    def send(self, message):
        print(f"[Email] {message}")
        return True


class SMSNotifier(Notifier):

    def send(self, message):
        print(f"[SMS] {message}")
        return True


class NotificationFactory(ABC):

    @abstractmethod
    def create_notifier(self) -> Notifier:
        ...

    def deliver(self, message: str) -> bool:  # 模板方法
        notifier = self.create_notifier()
        print("-> 前置校验通过")
        ok = notifier.send(message)
        print("-> 后置记录日志")
        return ok


class EmailFactory(NotificationFactory):

    def create_notifier(self):
        return EmailNotifier()

1
2
EmailFactory().deliver("您的验证码是 1234")

1
2
3
4
-> 前置校验通过
[Email] 您的验证码是 1234
-> 后置记录日志

这里叠了两层模式：Notifier 是抽象产品，NotificationFactory 是工厂方法（create_notifier 抽象），deliver 又是模板方法（流程固定、产品可变）。两层抽象各管一段职责，新增一个渠道只要加两个小类，不动任何既有代码。

第五部分 工程落地建议

契约要小。 抽象方法只放真正必须由每个实现提供的能力。可选项给默认实现。抽象面越大，”改契约”就越像”改公共 API”。

变更要当迁移处理。 往抽象基类里加一个抽象方法，等于让所有现有子类立刻报错。稳妥的顺序是：先加一个带默认实现的具体方法，或加一个能力探测方法（如 supports_batch_write()），再逐步收紧。

异常语义要写进契约。 明确哪些失败是返回值、哪些是异常、哪些可重试。成熟系统会定义领域异常层次，要求各实现把底层错误统一映射过来。

什么时候不要用 ABC。 只有一个内部小类、没有扩展计划；短脚本、测试已经覆盖了行为；需要靠结构化匹配适应已有的、不能改的类——这些情况下，鸭子类型或 Protocol 更合适。抽象基类不是越用越好，它是用来减少意外的。

测试上要覆盖”漏实现”。 写一个测试，断言某个故意漏实现的子类在实例化时抛 TypeError，等于给契约上了一把锁。

附录 速查表

需求 写法
定义抽象基类 class X(ABC):
抽象方法 @abstractmethod def m(self): ...
抽象属性 @property 在上，@abstractmethod 在下
抽象类方法 @classmethod 在上，@abstractmethod 在下
抽象静态方法 @staticmethod 在上，@abstractmethod 在下
抽象方法带默认实现 直接在方法体里写逻辑，子类用 super() 调用
查看未实现的抽象成员 Cls.__abstractmethods__（frozenset）
判断是否抽象类 inspect.isabstract(Cls)
注册虚拟子类 Base.register(Other)
结构化判定 重写 __subclasshook__（定义为 classmethod）
类创建后重算抽象状态 update_abstractmethods(Cls)（3.10+）
注册表缓存令牌 get_cache_token()

装饰器顺序口诀：角色装饰器在上，@abstractmethod 永远贴着 def。

结语

抽象基类解决的不是”怎么写代码”，而是”怎么让一堆代码不互相拆台”。它的三个收益很实在：把口头约定变成实例化时的硬约束，把接口文档固化成可执行的类，把公共流程收敛到一处。代价是类变多、层次变深，所以要按 3.4 的判据权衡——需要运行时把关就上 ABC，只需要静态检查就用 Protocol，范围可控就留给鸭子类型。

Python > Python 常见函数 > Python 抽象类
#原创 #python #ABC 抽象基类
pythonABC抽象模块
https://dreamshao.github.io/2026/09/29/pythonABC抽象模块/
作者
Yun Shao
发布于
2026年9月29日
许可协议