Python3 抽象基类 ABC:强制接口契约

🎉摘要:深入解析Python抽象基类(ABC)的用法,从鸭子类型的隐患出发,讲解如何用abc模块定义抽象方法、属性和类方法,利用虚拟子类注册和内置抽象基类(collections.abc)构建清晰接口,结合类型注解与插件框架实战,在实例化时拦截错误,写出不易误用的类层次结构。

鸭子类型(duck typing)是 Python 的优势,但也带来了隐患:一个对象声称支持“序列”协议,却没有 __getitem__ 方法,调用时就会崩溃。

抽象基类(Abstract Base Class,ABC)的用途正是在运行前或实例化时就拦截这种错误,强制子类实现约定的方法。

本章从 abc 模块讲起,覆盖抽象方法、抽象属性、抽象类方法,以及虚拟子类注册,让你写出接口清晰、不易误用的类层次结构。

为什么需要抽象基类?

鸭子类型的盲区

下面代码模拟没有 ABC 的情况下,出现的灾难问题。实际代码应该是这样:

class FakeFile:
    def read(self):
        # 模拟读取配置内容
        return '{"env": "test", "debug": true}'

def read_config(file_obj):
    return file_obj.read()

print(read_config(FakeFile()))

但是,在开发中,由于其他原因忘记在 FakeFile 下面实现 read() 方法了(如果在类似 Java 的语言中,编译都过不了,因为 read() 函数不存在):

class FakeFile:
    pass  # 假装是个文件,但什么方法都没实现

def read_config(file_obj):
    return file_obj.read()  # 运行时才会爆炸

read_config(FakeFile())  # AttributeError: 'FakeFile' object has no attribute 'read'

运行上面代码,出现如下错误:

Traceback (most recent call last):
  File "d:\python_demo\demo.py", line 7, in <module>
    read_config(FakeFile())  # AttributeError: 'FakeFile' object has no attribute 'read'
    ~~~~~~~~~~~^^^^^^^^^^^^
  File "d:\python_demo\demo.py", line 5, in read_config
    return file_obj.read()  # 运行时才会爆炸
           ^^^^^^^^^^^^^
AttributeError: 'FakeFile' object has no attribute 'read'

这个错误直到运行时才发现,意味着生产环境可能突然崩溃。ABC 可以把这种检查提前到实例化时,甚至在类定义时。

ABC 的核心价值

下面通过四个场景介绍 ABC 存在的价值:

场景没有 ABC有 ABC
子类忘记实现方法运行时报错实例化时直接拒绝,报错明确
接口文档靠注释和文档代码即契约,IDE 自动提示
多团队协作依赖口头约定抽象基类就是强制规范
第三方扩展不知道要重写什么未实现的方法一目了然

📌注意

ABC 不是让你回到 Java 的刻板类型系统,而是给 Python 的灵活性加一道安全网。它告诉你"如果你自称是这个类型,就必须具备这些能力"。

ABC 基础:定义与使用

最小抽象类

下面示例演示使用 ABC 定义一个最小抽小类,类中仅存在一个抽象方法:

from abc import ABC, abstractmethod

class BaseTask(ABC):
    """所有任务必须实现 run 方法"""

    @abstractmethod
    def run(self) -> None:
        """执行任务,子类必须重写"""
        pass

    def status(self) -> str:
        """可选方法,子类可直接继承"""
        return "idle"

# 尝试实例化抽象类
try:
    task = BaseTask() # 错误
except TypeError as e:
    print(f"拦截: {e}")
    # 输出: Can't instantiate abstract class BaseTask with abstract method run

运行上面代码将直接抛出错误,因为不能实例化抽象类(一个类中包含被标记为抽象方法,方法只有声明、没有具体实现,这类类就是抽象类)。

关键点:

  • 继承 ABC(或 abc.ABCMeta 元类)才能拥有抽象能力;

  • @abstractmethod 装饰的方法,即抽象方法,子类必须实现;

  • 抽象类可以有非抽象方法(status),子类直接继承使用;

  • 抽象类本身不能实例化。

子类实现抽象方法

示例演示继承上面的 BaseTask 类,实现 run() 方法:

from abc import ABC, abstractmethod

class BaseTask(ABC):
    """所有任务必须实现 run 方法"""

    @abstractmethod
    def run(self) -> None:
        """执行任务,子类必须重写"""
        pass

    def status(self) -> str:
        """可选方法,子类可直接继承"""
        return "idle"

class DownloadTask(BaseTask):
    """下载任务,继承自 BaseTask"""
    def __init__(self, url: str):
        self.url = url

    def run(self) -> None:
        print(f"Downloading {self.url}...")


# 现在可以实例化了
dl = DownloadTask("https://example.com/file.zip")
dl.run()

运行示例,输出如下:

Downloading https://example.com/file.zip...

📌注意

如果类没有重写 run,实例化时将直接抛出 TypeError。这比运行时才发现少了 AttributeError 的隐藏炸弹。

抽象属性:用 property 强制实现

方法可以抽象,属性也可以。abc 模块提供了 abstractproperty(已弃用)和 abstractmethod + property 的组合写法。例如:

from abc import ABC, abstractmethod

class Shape(ABC):
    @property
    @abstractmethod
    def area(self) -> float:
        """子类必须实现 area 属性"""
        pass

    @property
    @abstractmethod
    def perimeter(self) -> float:
        """子类必须实现 perimeter 属性"""
        pass

    def describe(self) -> str:
        return f"Area: {self.area}, Perimeter: {self.perimeter}"


class Rectangle(Shape):
    """矩形类,继承自 Shape"""
    def __init__(self, width: float, height: float):
        self.width = width
        self.height = height

    @property
    def area(self) -> float:
        return self.width * self.height

    @property
    def perimeter(self) -> float:
        return 2 * (self.width + self.height)

if __name__ == "__main__":
    rect = Rectangle(3, 4)
    print(rect.describe())  # Area: 12, Perimeter: 14

    # 如果忘记实现 area
    class BadShape(Shape):
        @property
        def perimeter(self) -> float:
            return 0

    try:
        BadShape()
    except TypeError as e:
        print(f"BadShape: {e}")
        # 提示未实现 abstract property 'area'

运行代码,输出如下:

Area: 12, Perimeter: 14
BadShape: Can't instantiate abstract class BadShape without an implementation for abstract method 'area'

和抽象类一样,如果子类没有实现抽象属性,实例化时也会直接报错。

注意,装饰器顺序:@property 必须写在 @abstractmethod 上面。如果顺序颠倒,abstractmethod 会看不到下面的 property,导致检查失效。例如:

# 错误!
class WrongOrder(ABC):
    @abstractmethod
    @property        # 顺序颠倒
    def value(self):
        pass

抽象类方法与抽象静态方法

类方法和静态方法同样可以标记为抽象。例如:

from abc import ABC, abstractmethod

class DatabaseDriver(ABC):
    @abstractmethod
    def connect(self, url: str) -> None:
        """实例方法:建立连接"""
        pass

    @classmethod
    @abstractmethod
    def driver_name(cls) -> str:
        """类方法:返回驱动名称"""
        pass

    @staticmethod
    @abstractmethod
    def supported_schemes() -> list[str]:
        """静态方法:返回支持的 URL scheme"""
        pass

class PostgresDriver(DatabaseDriver):
    def connect(self, url: str) -> None:
        print(f"Connecting to PostgreSQL: {url}")

    @classmethod
    def driver_name(cls) -> str:
        return "postgresql"

    @staticmethod
    def supported_schemes() -> list[str]:
        return ["postgres", "postgresql"]

pg = PostgresDriver()
pg.connect("postgres://localhost/mydb")
print(PostgresDriver.driver_name())
print(PostgresDriver.supported_schemes())

运行示例,输出如下:

Connecting to PostgreSQL: postgres://localhost/mydb
postgresql
['postgres', 'postgresql']

@classmethod 和 @staticmethod 的顺序也要注意:抽象装饰器应该在最外层。即 abstractmethod 包裹 classmethod。记住一条规则即可:@abstractmethod 总是紧贴方法定义。

虚拟子类:不用继承也能"声称"是子类

ABC 有一个强大的功能:你可以让一个类在不继承 ABC 的情况下,被注册为 ABC 的“虚拟子类”。这用于给已经存在的类(如第三方库或内置类型)打上接口标签。例如:

from abc import ABC, abstractmethod

class Reader(ABC):
    @abstractmethod
    def read(self, size: int = -1) -> bytes:
        pass

# 已有类,不想修改它的继承链
class MyBytesIO:
    def __init__(self, data: bytes):
        self.data = data
        self.pos = 0

    def read(self, size: int = -1) -> bytes:
        if size == -1:
            result = self.data[self.pos:]
            self.pos = len(self.data)
            return result
        result = self.data[self.pos:self.pos + size]
        self.pos += size
        return result

# 注册为虚拟子类
Reader.register(MyBytesIO)

print(issubclass(MyBytesIO, Reader))  # True
print(isinstance(MyBytesIO(b"hello"), Reader))  # True

# 验证 read 是否工作
r = MyBytesIO(b"hello world")
assert issubclass(type(r), Reader)
print(r.read(5))  # b'hello'

运行代码,输出如下:

True
True
b'hello'

register() 不会检查 MyBytesIO 是否真的实现了 read。它只是更新了 issubclass 和 isinstance 的返回结果。这是一个信任声明,由开发者保证被注册类确实遵守接口契约。

为什么需要虚拟子类?

假设你写了一个库,要求传入的对象必须是 Reader 类型。用户手里有一个 MyBytesIO,但它没继承 Reader。你可以让用户改源码,但更好的方式是用 register() 非侵入式地注册:

# 在库代码中,或者在用户入口代码里注册
from my_library import Reader
from some_third_party import MyBytesIO

Reader.register(MyBytesIO)

# 现在可以传入 MyBytesIO 实例了
library.process(io_instance)

这避免了强迫第三方库修改继承链,也保持了 Python 的鸭子类型传统。

内置抽象基类:collections.abc

Python 标准库的 collections.abc 模块定义了大量常用的抽象基类,如 Sequence、Mapping、Iterable 等。它们既是接口规范,也是类型检查工具。例如:

from collections.abc import Sequence, Mapping, Iterable, Sized, Container

# 判断内置类型是否属于这些 ABC
print(isinstance([1, 2, 3], Sequence))      # True
print(isinstance({"a": 1}, Mapping))        # True
print(isinstance("hello", Iterable))        # True
print(isinstance({1, 2}, Container))        # True
print(isinstance((1, 2, 3), Sized))         # True

# 检查必须实现的方法
print(Sequence.__abstractmethods__)  # {'__getitem__', '__len__'}
print(Mapping.__abstractmethods__)   # {'__getitem__', '__iter__', '__len__'}

collections.abc 的常用抽象基类:

ABC必须实现继承后自动获得
Iterable__iter____contains__(通过遍历)
Iterator__next____iter__
Sequence__getitem__, __len__index, count, __contains__
Mapping__getitem__, __iter__, __len__keys, values, items, get
MutableMapping__setitem__, __delitem__update, pop, setdefault
Set__contains__, __iter__, __len__issubset, union, intersection

自定义 Sequence

通过继承 collections.abc 模块的 Sequence,实现自己的 Sequence,例如:

from collections.abc import Sequence

class MyList(Sequence):
    def __init__(self, data):
        self._data = list(data)

    def __getitem__(self, index):
        return self._data[index]

    def __len__(self):
        return len(self._data)

ml = MyList([10, 20, 30])
print(ml[1])         # 20
print(len(ml))       # 3
print(20 in ml)      # True —— 继承自 Sequence
print(ml.index(20))  # 1 —— 继承自 Sequence
print(ml.count(10))  # 1 —— 继承自 Sequence

只要实现了 __getitem__ 和 __len__,Sequence 会自动提供 index、count、__contains__ 等方法。这比从零手写一个序列类省力得多,也保证了行为与内置序列一致。

自定义 Mapping

和自定义 Sequence 一样,也通过继承 collections.abc 模块的 Mapping 抽象类:

from collections.abc import Mapping

class ReadOnlyDict(Mapping):
    def __init__(self, data):
        self._data = dict(data)

    def __getitem__(self, key):
        return self._data[key]

    def __iter__(self):
        return iter(self._data)

    def __len__(self):
        return len(self._data)

rod = ReadOnlyDict({"a": 1, "b": 2})
print(list(rod.keys()))     # ['a', 'b']
print(list(rod.values()))   # [1, 2]
print(rod.get("c", 0))      # 0 —— 继承自 Mapping

# 不支持赋值,因为不是 MutableMapping
# rod["c"] = 3  # TypeError

抽象基类与类型注解

Python 3.5+ 的类型提示系统(typing 模块)和 ABC 是互补关系。typing 提供静态类型检查,abc 提供运行时检查。两者结合,可以让接口既被类型检查器理解,又在运行时被强制。例如:

from abc import ABC, abstractmethod
from typing import List, Dict, Any

class DataProcessor(ABC):
    @abstractmethod
    def load(self, source: str) -> List[Dict[str, Any]]:
        """从 source 加载数据,返回记录列表"""
        pass

    @abstractmethod
    def transform(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """转换数据"""
        pass

    def run_pipeline(self, source: str) -> List[Dict[str, Any]]:
        """模板方法:子类只需实现 load 和 transform"""
        raw = self.load(source)
        return self.transform(raw)

class CsvProcessor(DataProcessor):
    def load(self, source: str) -> List[Dict[str, Any]]:
        # 简化实现
        return [{"col1": 1, "col2": 2}]

    def transform(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        return [{k.upper(): v for k, v in r.items()} for r in records]

p = CsvProcessor()
print(p.run_pipeline("data.csv"))  # [{'COL1': 1, 'COL2': 2}]

run_pipeline 是模板方法:父类定义流程骨架,子类填充具体步骤。这是 ABC 的经典设计模式,配合类型注解后,IDE 能自动提示子类需要实现哪些方法,参数类型一目了然。

综合实战:插件框架

把 ABC 的所有特性串起来,设计一个支持多种存储后端的插件框架。要求:每个插件必须实现连接、读取、写入、关闭;框架自动注册插件,并按优先级加载。

from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional

class StoragePlugin(ABC):
    """存储插件抽象基类"""

    @classmethod
    @abstractmethod
    def name(cls) -> str:
        """插件名称"""
        pass

    @classmethod
    @abstractmethod
    def priority(cls) -> int:
        """优先级,越高越优先使用"""
        pass

    @abstractmethod
    def connect(self, config: Dict[str, Any]) -> None:
        """连接存储"""
        pass

    @abstractmethod
    def read(self, key: str) -> Optional[Any]:
        """读取数据"""
        pass

    @abstractmethod
    def write(self, key: str, value: Any) -> None:
        """写入数据"""
        pass

    @abstractmethod
    def close(self) -> None:
        """关闭连接"""
        pass

    def __enter__(self):
        # 确保在进入上下文管理器时连接存储
        return self

    def __exit__(self, *args):
        # 确保在退出上下文管理器时关闭连接
        self.close()


class MemoryStorage(StoragePlugin):
    """内存存储插件"""

    @classmethod
    def name(cls) -> str:
        return "memory"

    @classmethod
    def priority(cls) -> int:
        return 10

    def __init__(self):
        self._data = {}
        self._connected = False

    def connect(self, config: Dict[str, Any]) -> None:
        self._connected = True
        print(f"[MemoryStorage] Connected")

    def read(self, key: str) -> Optional[Any]:
        return self._data.get(key)

    def write(self, key: str, value: Any) -> None:
        self._data[key] = value

    def close(self) -> None:
        self._data.clear()
        self._connected = False
        print(f"[MemoryStorage] Closed")


class FileStorage(StoragePlugin):
    """文件存储插件"""

    @classmethod
    def name(cls) -> str:
        return "file"

    @classmethod
    def priority(cls) -> int:
        return 5

    def __init__(self):
        self._path = None
        self._fp = None

    def connect(self, config: Dict[str, Any]) -> None:
        self._path = config.get("path", "/tmp/storage.json")
        print(f"[FileStorage] Connected to {self._path}")

    def read(self, key: str) -> Optional[Any]:
        import json
        try:
            with open(self._path, 'r') as f:
                data = json.load(f)
                return data.get(key)
        except FileNotFoundError:
            return None

    def write(self, key: str, value: Any) -> None:
        import json
        try:
            with open(self._path, 'r') as f:
                data = json.load(f)
        except FileNotFoundError:
            data = {}
        data[key] = value
        with open(self._path, 'w') as f:
            json.dump(data, f)

    def close(self) -> None:
        print(f"[FileStorage] Closed")


# 插件管理器
class PluginManager:
    def __init__(self):
        self._plugins: List[type[StoragePlugin]] = []

    def register(self, plugin_cls: type[StoragePlugin]) -> None:
        if not issubclass(plugin_cls, StoragePlugin):
            raise TypeError(f"{plugin_cls} 不是 StoragePlugin 的子类")
        self._plugins.append(plugin_cls)
        self._plugins.sort(key=lambda p: p.priority(), reverse=True)

    def create(self, name: str, config: Dict[str, Any]) -> Optional[StoragePlugin]:
        for plugin_cls in self._plugins:
            if plugin_cls.name() == name:
                instance = plugin_cls()
                instance.connect(config)
                return instance
        return None

    def list_plugins(self) -> List[str]:
        return [p.name() for p in self._plugins]


# 使用
if __name__ == "__main__":
    manager = PluginManager()
    manager.register(MemoryStorage)
    manager.register(FileStorage)

    print(f"已注册插件: {manager.list_plugins()}")

    with manager.create("memory", {}) as storage:
        storage.write("user_1", {"name": "Alice"})
        print(storage.read("user_1"))

    # 尝试注册不完整的插件
    try:
        class BadPlugin(StoragePlugin):
            pass
        manager.register(BadPlugin)
    except TypeError as e:
        print(f"BadPlugin 无法注册: {e}")

运行代码,输出如下:

已注册插件: ['memory', 'file']
[MemoryStorage] Connected
{'name': 'Alice'}
[MemoryStorage] Closed
BadPlugin 无法注册: Can't instantiate abstract class BadPlugin with abstract methods close, connect, name, priority, read, write

常见陷阱

抽象方法里该写 pass 还是 raise NotImplementedError?

class Base(ABC):
    @abstractmethod
    def method(self):
        pass  # 推荐,简洁,ABC 强制子类实现

class Base2:
    def method(self):
        raise NotImplementedError("子类必须实现 method")  # 没有 ABC 时的老做法

如果用了 ABC,pass 就够了,因为 @abstractmethod 已经负责强制实现。如果不用 ABC(比如旧代码或简单脚本),用 raise NotImplementedError 更明确。不要同时用两者,那是冗余。

ABC 与 Protocol 的选择

Python 3.8+ 引入了 typing.Protocol,它也是一种接口定义,但不需要显式继承:

from typing import Protocol


class Readable(Protocol):
    """
    结构化协议(鸭子类型校验)
    只要类实现签名一致的 read 方法,无需显式继承,自动满足该协议
    """
    def read(self, size: int = -1) -> bytes:
        """
        读取二进制数据
        :param size: 读取字节长度,-1 代表读取全部数据
        :return: 二进制字节串 bytes
        """
        ...


def process(reader: Readable) -> None:
    """
    统一处理任意实现 Readable 协议的可读二进制流对象
    :param reader: 满足 Readable 协议的二进制读取对象
    """
    # 读取最多100字节二进制数据
    data: bytes = reader.read(100)
    print(f"本次读取字节数:{len(data)}")
    print(f"读取原始数据:{data}")

    # 扩展简单业务逻辑示例
    if data:
        # 简单转字符串打印(utf-8解码)
        try:
            text = data.decode("utf-8")
            print(f"解码文本内容:{text}\n")
        except UnicodeDecodeError:
            print("数据非UTF-8文本,跳过解码\n")
    else:
        print("未读取到任何数据,数据流已读完\n")


# 测试场景:Python内置文件二进制流(天然符合协议)
def test_file_stream():
    # 以二进制只读模式打开文件,内置文件对象自带 read(size: int)->bytes,自动匹配Readable协议
    with open("temp_test.txt", "wb+") as f:
        f.write("Hello Protocol Test 中文测试".encode("utf-8"))
        f.seek(0)
        process(f)


# 测试场景:自定义类,隐式满足协议(无需继承)
class MockBinaryReader:
    """模拟二进制读取器,只要方法签名匹配协议,就被类型识别为Readable"""
    def __init__(self, content: bytes):
        self._buffer = content
        self._pos = 0

    def read(self, size: int = -1) -> bytes:
        if size == -1:
            res = self._buffer[self._pos:]
            self._pos = len(self._buffer)
        else:
            end = self._pos + size
            res = self._buffer[self._pos:end]
            self._pos = end
        return res


# 测试场景:不满足协议的对象调用会被类型检查工具拦截
class InvalidReader:
    # 返回str而非bytes,签名不匹配协议
    def read(self, size: int = -1) -> str:
        return "abc"


if __name__ == "__main__":
    test_file_stream()
    
    mock_obj = MockBinaryReader(b"1234567890_abcdefghij")
    process(mock_obj)
    
    process(InvalidReader())  # mypy/pylance静态类型检测直接报错

Protocol 是结构化的鸭子类型,只被静态类型检查器(mypy)识别,运行时没有强制力。ABC 是名义化类型,需要显式继承,运行时会强制实现。两者的选择:

  • 如果只需要 IDE 提示和 mypy 检查,用 Protocol 更灵活;

  • 如果需要运行时强制拦截,用 ABC;

  • 两者可以并存:类显式继承 ABC,同时参数类型用 Protocol。

说说我的看法
全部评论(
没有评论
关于
本网站专注于 Java、数据库(MySQL、Oracle)、Linux、软件架构及大数据等多领域技术知识分享。涵盖丰富的原创与精选技术文章,助力技术传播与交流。无论是技术新手渴望入门,还是资深开发者寻求进阶,这里都能为您提供深度见解与实用经验,让复杂编码变得轻松易懂,携手共赴技术提升新高度。如有侵权,请来信告知:hxstrive@outlook.com
其他应用
公众号