抽象基类与abc模块
抽象基类与abc模块
抽象基类(Abstract Base Classes,ABC)是Python中用于定义接口规范的重要工具。通过abc模块,开发者可以创建不能被实例化的基类,强制子类实现特定的方法,从而确保接口的一致性和完整性。
抽象基类基础
抽象基类使用abc.ABC作为基类,通过@abstractmethod装饰器定义抽象方法。子类必须实现所有抽象方法才能被实例化。
from abc import ABC, abstractmethod
import math
# 定义抽象基类
class Shape(ABC):
@abstractmethod
def area(self):
"""计算面积"""
pass
@abstractmethod
def perimeter(self):
"""计算周长"""
pass
def describe(self):
"""非抽象方法,提供默认实现"""
return f"{self.__class__.__name__}: 面积={self.area():.2f}, 周长={self.perimeter():.2f}"
# 具体实现类
class Circle(Shape):
def __init__(self, radius):
self.radius = radius
def area(self):
return math.pi * self.radius ** 2
def perimeter(self):
return 2 * math.pi * self.radius
class Rectangle(Shape):
def __init__(self, width, height):
self.width = width
self.height = height
def area(self):
return self.width * self.height
def perimeter(self):
return 2 * (self.width + self.height)
# 测试抽象基类
print("抽象基类测试:")
try:
# 尝试实例化抽象类
shape = Shape()
except TypeError as e:
print(f"无法实例化抽象类: {e}")
# 实例化具体类
circle = Circle(5)
rectangle = Rectangle(4, 6)
print(f"圆形: {circle.describe()}")
print(f"矩形: {rectangle.describe()}")
# 验证抽象方法实现
print(f"\n抽象方法实现验证:")
print(f"Circle实现了area: {hasattr(Circle, 'area') and not getattr(Circle.area, '__isabstractmethod__', False)}")
print(f"Rectangle实现了perimeter: {hasattr(Rectangle, 'perimeter') and not getattr(Rectangle.perimeter, '__isabstractmethod__', False)}")
抽象属性与静态抽象方法
除了抽象方法,abc模块还支持抽象属性、类方法和静态方法,为接口设计提供更多灵活性。
from abc import ABC, abstractmethod
import math
# 抽象属性示例
class Vehicle(ABC):
@property
@abstractmethod
def speed(self):
"""抽象属性:速度"""
pass
@abstractmethod
def accelerate(self, amount):
"""抽象方法:加速"""
pass
@classmethod
@abstractmethod
def create_default(cls):
"""抽象类方法:创建默认实例"""
pass
@staticmethod
@abstractmethod
def fuel_efficiency():
"""抽象静态方法:燃油效率"""
pass
class Car(Vehicle):
def __init__(self, speed=0):
self._speed = speed
@property
def speed(self):
return self._speed
def accelerate(self, amount):
self._speed += amount
@classmethod
def create_default(cls):
return cls(speed=60)
@staticmethod
def fuel_efficiency():
return 15.5 # 公里/升
# 测试抽象属性和方法
car = Car.create_default()
print(f"默认汽车速度: {car.speed}")
car.accelerate(20)
print(f"加速后速度: {car.speed}")
print(f"燃油效率: {car.fuel_efficiency()} 升/公里")
# 抽象属性验证
print(f"\n抽象属性验证:")
print(f"Car.speed是属性: {isinstance(Car.__dict__['speed'], property)}")
print(f"speed有abstractmethod装饰器: {getattr(Car.speed.fget, '__isabstractmethod__', False)}")
多重继承与Mixin
抽象基类支持多重继承,可以创建Mixin类来组合不同的功能接口,实现灵活的代码复用。
from abc import ABC, abstractmethod
from typing import List, Dict, Any
# Mixin抽象基类
class Readable(ABC):
@abstractmethod
def read(self) -> str:
pass
@abstractmethod
def readlines(self) -> List[str]:
pass
class Writable(ABC):
@abstractmethod
def write(self, data: str) -> None:
pass
@abstractmethod
def writeln(self, line: str) -> None:
pass
class Seekable(ABC):
@abstractmethod
def seek(self, position: int) -> None:
pass
@abstractmethod
def tell(self) -> int:
pass
# 组合多重抽象基类
class FileHandler(Readable, Writable, Seekable):
def __init__(self, filename):
self.filename = filename
self._position = 0
self._content = ""
def read(self) -> str:
return self._content
def readlines(self) -> List[str]:
return self._content.split('\n')
def write(self, data: str) -> None:
self._content += data
def writeln(self, line: str) -> None:
self._content += line + '\n'
def seek(self, position: int) -> None:
self._position = position
def tell(self) -> int:
return self._position
# 测试多重继承
handler = FileHandler("test.txt")
handler.writeln("第一行")
handler.writeln("第二行")
handler.write("第三行")
print(f"文件内容: {handler.read()}")
print(f"行数: {len(handler.readlines())}")
handler.seek(5)
print(f"当前位置: {handler.tell()}")
# 验证接口完整性
print(f"\n接口完整性验证:")
print(f"FileHandler实现了Readable: {issubclass(FileHandler, Readable)}")
print(f"FileHandler实现了Writable: {issubclass(FileHandler, Writable)}")
print(f"FileHandler实现了Seekable: {issubclass(FileHandler, Seekable)}")
注册虚拟子类
abc模块允许通过register方法注册虚拟子类,无需显式继承即可使类被视为抽象基类的子类,这在适配器模式中非常有用。
from abc import ABC, abstractmethod
from typing import Sequence, Iterator
# 定义抽象基类
class Database(ABC):
@abstractmethod
def connect(self, connection_string: str) -> None:
pass
@abstractmethod
def execute(self, query: str) -> Any:
pass
@abstractmethod
def close(self) -> None:
pass
# 第三方数据库类(没有继承Database)
class LegacyDatabase:
def __init__(self):
self.connected = False
def open_connection(self, conn_str):
self.connected = True
print(f"Legacy数据库连接: {conn_str}")
def run_query(self, sql):
return f"Legacy查询结果: {sql}"
def disconnect(self):
self.connected = False
print("Legacy数据库断开")
# 适配器类
class DatabaseAdapter(Database):
def __init__(self, legacy_db):
self.legacy_db = legacy_db
def connect(self, connection_string: str) -> None:
self.legacy_db.open_connection(connection_string)
def execute(self, query: str) -> Any:
return self.legacy_db.run_query(query)
def close(self) -> None:
self.legacy_db.disconnect()
# 注册虚拟子类
Database.register(LegacyDatabase)
# 测试虚拟子类注册
legacy_db = LegacyDatabase()
print(f"LegacyDatabase是Database的子类: {isinstance(legacy_db, Database)}")
# 使用适配器模式
adapter = DatabaseAdapter(legacy_db)
adapter.connect("localhost:5432")
result = adapter.execute("SELECT * FROM users")
print(f"查询结果: {result}")
adapter.close()
# 自定义序列类型
class CustomSequence(Sequence):
def __init__(self, data):
self._data = list(data)
def __getitem__(self, index):
return self._data[index]
def __len__(self):
return len(self._data)
# 注册虚拟子类
Sequence.register(CustomSequence)
custom_seq = CustomSequence([1, 2, 3, 4, 5])
print(f"\n自定义序列: {custom_seq}")
print(f"序列长度: {len(custom_seq)}")
print(f"第一个元素: {custom_seq[0]}")
print(f"是Sequence实例: {isinstance(custom_seq, Sequence)}")
实际应用:插件系统设计
抽象基类非常适合设计插件系统,确保所有插件都实现必要的接口,同时保持灵活性。
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any
import json
# 插件接口定义
class PluginInterface(ABC):
@property
@abstractmethod
def name(self) -> str:
"""插件名称"""
pass
@property
@abstractmethod
def version(self) -> str:
"""插件版本"""
pass
@abstractmethod
def initialize(self, config: Dict[str, Any]) -> bool:
"""初始化插件"""
pass
@abstractmethod
def execute(self, data: Any) -> Any:
"""执行插件功能"""
pass
@abstractmethod
def cleanup(self) -> None:
"""清理资源"""
pass
# 具体插件实现
class DataValidatorPlugin(PluginInterface):
@property
def name(self) -> str:
return "数据验证插件"
@property
def version(self) -> str:
return "1.0.0"
def initialize(self, config: Dict[str, Any]) -> bool:
self.rules = config.get('rules', {})
return True
def execute(self, data: Any) -> Any:
errors = []
for field, rule in self.rules.items():
if field in data:
if not self._validate_field(data[field], rule):
errors.append(f"字段 {field} 验证失败")
return {"valid": len(errors) == 0, "errors": errors}
def _validate_field(self, value, rule):
if 'type' in rule and not isinstance(value, rule['type']):
return False
if 'min' in rule and value < rule['min']:
return False
if 'max' in rule and value > rule['max']:
return False
return True
def cleanup(self) -> None:
self.rules = {}
class DataTransformerPlugin(PluginInterface):
@property
def name(self) -> str:
return "数据转换插件"
@property
def version(self) -> str:
return "1.0.0"
def initialize(self, config: Dict[str, Any]) -> bool:
self.transformations = config.get('transformations', {})
return True
def execute(self, data: Any) -> Any:
result = data.copy()
for field, transform in self.transformations.items():
if field in result:
result[field] = self._apply_transform(result[field], transform)
return result
def _apply_transform(self, value, transform):
if transform == 'upper':
return str(value).upper()
elif transform == 'lower':
return str(value).lower()
elif transform == 'strip':
return str(value).strip()
return value
def cleanup(self) -> None:
self.transformations = {}
# 插件管理器
class PluginManager:
def __init__(self):
self.plugins: Dict[str, PluginInterface] = {}
def register_plugin(self, plugin: PluginInterface) -> bool:
"""注册插件"""
try:
if not isinstance(plugin, PluginInterface):
raise TypeError("插件必须实现PluginInterface接口")
self.plugins[plugin.name] = plugin
print(f"插件注册成功: {plugin.name} v{plugin.version}")
return True
except Exception as e:
print(f"插件注册失败: {e}")
return False
def initialize_plugin(self, name: str, config: Dict[str, Any]) -> bool:
"""初始化插件"""
if name not in self.plugins:
print(f"插件不存在: {name}")
return False
return self.plugins[name].initialize(config)
def execute_plugin(self, name: str, data: Any) -> Any:
"""执行插件"""
if name not in self.plugins:
raise ValueError(f"插件不存在: {name}")
return self.plugins[name].execute(data)
def cleanup_all(self):
"""清理所有插件"""
for plugin in self.plugins.values():
plugin.cleanup()
print("所有插件已清理")
# 测试插件系统
manager = PluginManager()
# 注册插件
validator = DataValidatorPlugin()
transformer = DataTransformerPlugin()
manager.register_plugin(validator)
manager.register_plugin(transformer)
# 初始化插件
validator_config = {
'rules': {
'age': {'type': int, 'min': 0, 'max': 150},
'name': {'type': str, 'min': 1}
}
}
transformer_config = {
'transformations': {
'name': 'upper',
'email': 'lower'
}
}
manager.initialize_plugin("数据验证插件", validator_config)
manager.initialize_plugin("数据转换插件", transformer_config)
# 使用插件
test_data = {
'name': ' John Doe ',
'email': 'JOHN@EXAMPLE.COM',
'age': 25
}
# 验证数据
validation_result = manager.execute_plugin("数据验证插件", test_data)
print(f"验证结果: {validation_result}")
# 转换数据
transformed_data = manager.execute_plugin("数据转换插件", test_data)
print(f"转换结果: {transformed_data}")
# 清理
manager.cleanup_all()
抽象基类和abc模块为Python提供了强大的接口定义和强制实现机制。通过合理使用抽象基类,开发者可以设计清晰的接口规范,实现代码的多态性和可扩展性,构建健壮的插件系统和框架。