← 返回首页
🐍

抽象基类与abc模块

📂 python ⏱ 5 min 896 words

抽象基类与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提供了强大的接口定义和强制实现机制。通过合理使用抽象基类,开发者可以设计清晰的接口规范,实现代码的多态性和可扩展性,构建健壮的插件系统和框架。