鸭子类型(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 |
| 子类忘记实现方法 | 运行时报错 | 实例化时直接拒绝,报错明确 |
| 接口文档 | 靠注释和文档 | 代码即契约,IDE 自动提示 |
| 多团队协作 | 依赖口头约定 | 抽象基类就是强制规范 |
| 第三方扩展 | 不知道要重写什么 | 未实现的方法一目了然 |
📌注意
ABC 不是让你回到 Java 的刻板类型系统,而是给 Python 的灵活性加一道安全网。它告诉你"如果你自称是这个类型,就必须具备这些能力"。
下面示例演示使用 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 的隐藏炸弹。
方法可以抽象,属性也可以。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 的鸭子类型传统。
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 |
通过继承 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__ 等方法。这比从零手写一个序列类省力得多,也保证了行为与内置序列一致。
和自定义 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 # TypeErrorPython 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, writeclass Base(ABC):
@abstractmethod
def method(self):
pass # 推荐,简洁,ABC 强制子类实现
class Base2:
def method(self):
raise NotImplementedError("子类必须实现 method") # 没有 ABC 时的老做法如果用了 ABC,pass 就够了,因为 @abstractmethod 已经负责强制实现。如果不用 ABC(比如旧代码或简单脚本),用 raise NotImplementedError 更明确。不要同时用两者,那是冗余。
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。