Python ABC 与元类编程：从协议定义到动态类型系统

1
from abc import ABC, abstractmethod
2

3
class Quackable(ABC):
4
    """一个会叫的协议"""
5

6
    @abstractmethod
7
    def quack(self) -> str:
8
        """发出叫声"""
9
        ...
10

11
class Duck(Quackable):
12
    def quack(self) -> str:
13
        return "Quack!"
14

15
# class Person(Quackable):
16
#     pass  # TypeError: Can't instantiate abstract class Person
17
#           # with abstract method quack

1
from abc import ABC, abstractmethod
2

3
class JSONSerializable(ABC):
4
    @abstractmethod
5
    def to_json(self) -> str:
6
        ...
7

8
class User:
9
    def __init__(self, name: str, age: int):
10
        self.name = name
11
        self.age = age
12

13
    def to_json(self) -> str:
14
        import json
15
        return json.dumps({"name": self.name, "age": self.age})
16

17
# 注册为虚拟子类（不修改 User 的 __bases__）
18
JSONSerializable.register(User)
19

20
user = User("Alice", 30)
21
print(isinstance(user, JSONSerializable))  # True
22
print(issubclass(User, JSONSerializable))  # True

1
from abc import ABC, abstractmethod
2

3
class DataSource(ABC):
4
    """数据源协议"""
5

6
    @property
7
    @abstractmethod
8
    def name(self) -> str:
9
        """数据源名称"""
10
        ...
11

12
    @classmethod
13
    @abstractmethod
14
    def from_config(cls, config: dict) -> "DataSource":
15
        """从配置创建实例"""
16
        ...
17

18
    @staticmethod
19
    @abstractmethod
20
    def validate_config(config: dict) -> bool:
21
        """验证配置是否合法"""
22
        ...
23

24
    @abstractmethod
25
    def query(self, sql: str) -> list[dict]:
26
        """执行查询"""
27
        ...
28

29
class MySQLSource(DataSource):
30
    def __init__(self, host: str, database: str):
31
        self._host = host
32
        self._database = database
33

34
    @property
35
    def name(self) -> str:
36
        return f"mysql://{self._host}/{self._database}"
37

38
    @classmethod
39
    def from_config(cls, config: dict) -> "MySQLSource":
40
        return cls(host=config["host"], database=config["database"])
41

42
    @staticmethod
43
    def validate_config(config: dict) -> bool:
44
        return "host" in config and "database" in config
45

46
    def query(self, sql: str) -> list[dict]:
47
        # 实际实现...
48
        return []

1
from abc import ABC, abstractmethod
2

3
class SizedIterable(ABC):
4
    @abstractmethod
5
    def __len__(self) -> int:
6
        ...
7

8
    @abstractmethod
9
    def __iter__(self):
10
        ...
11

12
    @classmethod
13
    def __subclasshook__(cls, subclass):
14
        # 只要类有 __len__ 和 __iter__，就认为是 SizedIterable 的子类
15
        if cls is SizedIterable:
16
            has_len = any("__len__" in B.__dict__ for B in subclass.__mro__)
17
            has_iter = any("__iter__" in B.__dict__ for B in subclass.__mro__)
18
            if has_len and has_iter:
19
                return True
20
        return NotImplemented
21

22
class MyCollection:
23
    def __len__(self):
24
        return 10
25
    def __iter__(self):
26
        return iter(range(10))
27

28
print(issubclass(MyCollection, SizedIterable))  # True
29
print(isinstance(MyCollection(), SizedIterable))  # True

1
class Meta(type):
2
    def __new__(mcs, name, bases, namespace, **kwargs):
3
        """创建类对象时调用（在类存在之前）
4

5
        参数：
6
          mcs: 元类本身
7
          name: 类名
8
          bases: 父类元组
9
          namespace: 类体中的名称空间（方法、属性等）
10
        """
11
        print(f"[Meta.__new__] 创建类: {name}")
12
        return super().__new__(mcs, name, bases, namespace)
13

14
    def __init__(cls, name, bases, namespace, **kwargs):
15
        """类对象创建后调用（类已存在）
16

17
        参数：
18
          cls: 刚创建的类对象
19
        """
20
        print(f"[Meta.__init__] 初始化类: {name}")
21
        super().__init__(name, bases, namespace)
22

23
    def __call__(cls, *args, **kwargs):
24
        """用类创建实例时调用（即 obj = MyClass() 时）
25

26
        这是 __new__ 和 __init__ 的上游拦截点。
27
        """
28
        print(f"[Meta.__call__] 创建 {cls.__name__} 实例, args={args}")
29
        return super().__call__(*args, **kwargs)
30

31
class MyClass(metaclass=Meta):
32
    def __init__(self, value):
33
        self.value = value
34
        print(f"[MyClass.__init__] value={value}")
35

36
print("--- 定义阶段 ---")
37
# 输出：
38
# [Meta.__new__] 创建类: MyClass
39
# [Meta.__init__] 初始化类: MyClass
40

41
print("\n--- 实例化阶段 ---")
42
obj = MyClass(42)
43
# 输出：
44
# [Meta.__call__] 创建 MyClass 实例, args=(42,)
45
# [MyClass.__init__] value=42

1
class PluginRegistry(type):
2
    """插件注册元类"""
3
    _plugins: dict[str, type] = {}
4

5
    def __new__(mcs, name, bases, namespace):
6
        cls = super().__new__(mcs, name, bases, namespace)
7
        # 跳过基类本身的注册
8
        if name != "BasePlugin":
9
            plugin_name = namespace.get("plugin_name", name.lower())
10
            PluginRegistry._plugins[plugin_name] = cls
11
        return cls
12

13
    @classmethod
14
    def get_plugin(mcs, name: str) -> type:
15
        if name not in mcs._plugins:
16
            available = ", ".join(mcs._plugins.keys())
17
            raise KeyError(f"未知插件: {name}。可用插件: {available}")
18
        return mcs._plugins[name]
19

20
    @classmethod
21
    def list_plugins(mcs) -> list[str]:
22
        return sorted(mcs._plugins.keys())
23

24
class BasePlugin(metaclass=PluginRegistry):
25
    """插件基类"""
26
    plugin_name: str = ""
27

28
    def execute(self) -> str:
29
        raise NotImplementedError
30

31
# 定义插件时自动注册
32
class ImageCompressPlugin(BasePlugin):
33
    plugin_name = "image_compress"
34

35
    def execute(self) -> str:
36
        return "Compressing images..."
37

38
class LogRotatePlugin(BasePlugin):
39
    plugin_name = "log_rotate"
40

41
    def execute(self) -> str:
42
        return "Rotating logs..."
43

44
class CacheCleanupPlugin(BasePlugin):
45
    plugin_name = "cache_cleanup"
46

47
    def execute(self) -> str:
48
        return "Cleaning up cache..."
49

50
# 使用注册表
51
print("已注册插件:", PluginRegistry.list_plugins())
52
# ['cache_cleanup', 'image_compress', 'log_rotate']
53

54
plugin_cls = PluginRegistry.get_plugin("image_compress")
55
plugin = plugin_cls()
56
print(plugin.execute())  # Compressing images...

1
class MetaA(type):
2
    pass
3

4
class MetaB(type):
5
    pass
6

7
class A(metaclass=MetaA):
8
    pass
9

10
class B(metaclass=MetaB):
11
    pass
12

13
# class C(A, B):
14
#     pass
15
# TypeError: metaclass conflict: the metaclass of A is MetaA,
16
#            the metaclass of B is MetaB

1
class BasePlugin:
2
    _registry: dict[str, type] = {}
3
    plugin_name: str = ""
4

5
    def __init_subclass__(cls, /, plugin_name: str = None, **kwargs):
6
        super().__init_subclass__(**kwargs)
7
        name = plugin_name or cls.plugin_name or cls.__name__.lower()
8
        cls.plugin_name = name
9
        if name not in ("baseplugin",):
10
            BasePlugin._registry[name] = cls
11
            print(f"✅ 注册插件: {name}")
12

13
    @classmethod
14
    def get_plugin(cls, name: str) -> type:
15
        return cls._registry[name]
16

17
class ImagePlugin(BasePlugin, plugin_name="image"):
18
    def process(self):
19
        return "Processing image"
20

21
class AudioPlugin(BasePlugin, plugin_name="audio"):
22
    def process(self):
23
        return "Processing audio"
24

25
print("注册表:", BasePlugin._registry)
26
# {'image': <class 'ImagePlugin'>, 'audio': <class 'AudioPlugin'>}

1
from typing import TypeVar, Generic
2

3
T = TypeVar("T")
4

5
class Repository(Generic[T]):
6
    """泛型仓库"""
7

8
    def save(self, entity: T) -> None:
9
        ...
10

11
    def find_by_id(self, entity_id: str) -> T | None:
12
        ...
13

14
# 类型标注
15
class UserRepository(Repository["User"]):
16
    def save(self, entity: "User") -> None:
17
        print(f"Saving user: {entity.name}")

1
from abc import ABC, abstractmethod
2
from typing import Any, ClassVar
3
import json
4

5
class Field:
6
    """字段描述符"""
7
    def __init__(self, field_type: type, required: bool = True, default: Any = None):
8
        self.field_type = field_type
9
        self.required = required
10
        self.default = default
11
        self.name = ""  # 由元类填充
12

13
    def __set_name__(self, owner, name):
14
        self.name = name
15

16
    def __set__(self, obj, value):
17
        if value is None:
18
            if self.required:
19
                raise ValueError(f"{self.name} 不能为 None")
20
            obj.__dict__[self.name] = self.default
21
            return
22

23
        if not isinstance(value, self.field_type):
24
            raise TypeError(
25
                f"{self.name} 期望 {self.field_type.__name__}, "
26
                f"得到 {type(value).__name__}"
27
            )
28
        obj.__dict__[self.name] = value
29

30
    def __get__(self, obj, objtype=None):
31
        if obj is None:
32
            return self
33
        return obj.__dict__.get(self.name, self.default)
34

35
class Validatable(ABC):
36
    """可验证对象协议"""
37
    @abstractmethod
38
    def validate(self) -> list[str]:
39
        """返回验证错误列表，空列表表示通过"""
40
        ...
41

42
class ModelMeta(type):
43
    """模型元类：收集所有 Field 并生成 __repr__"""
44
    def __new__(mcs, name, bases, namespace):
45
        # 收集字段
46
        fields = {}
47
        for attr_name, attr_value in namespace.items():
48
            if isinstance(attr_value, Field):
49
                fields[attr_name] = attr_value
50

51
        namespace["__fields__"] = fields
52
        namespace["__model_name__"] = name
53

54
        cls = super().__new__(mcs, name, bases, namespace)
55
        return cls
56

57
class Model(metaclass=ModelMeta):
58
    """模型基类"""
59
    __fields__: ClassVar[dict[str, Field]]
60
    __model_name__: ClassVar[str]
61

62
    def __init__(self, **kwargs):
63
        for field_name, field in self.__fields__.items():
64
            value = kwargs.get(field_name, field.default)
65
            if value is not None:
66
                setattr(self, field_name, value)
67
            elif field.required:
68
                raise ValueError(f"缺少必需字段: {field_name}")
69

70
    def __repr__(self) -> str:
71
        fields_str = ", ".join(
72
            f"{name}={getattr(self, name, '!')!r}"
73
            for name in self.__fields__
74
        )
75
        return f"{self.__model_name__}({fields_str})"
76

77
    def to_dict(self) -> dict:
78
        return {
79
            name: getattr(self, name, field.default)
80
            for name, field in self.__fields__.items()
81
        }
82

83
    def to_json(self) -> str:
84
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)
85

86
    @classmethod
87
    def from_json(cls, json_str: str) -> "Model":
88
        data = json.loads(json_str)
89
        return cls(**data)
90

91
    def validate(self) -> list[str]:
92
        """基础验证：检查所有必需字段"""
93
        errors = []
94
        for name, field in self.__fields__.items():
95
            value = getattr(self, name, None)
96
            if value is None and field.required:
97
                errors.append(f"{name} 不能为空")
98
        return errors

1
class User(Model, Validatable):
2
    name = Field(str, required=True)
3
    age = Field(int, required=True)
4
    email = Field(str, required=False, default="")
5

6
    def validate(self) -> list[str]:
7
        errors = super().validate()
8
        if self.age is not None and (self.age < 0 or self.age > 150):
9
            errors.append("age 必须在 0-150 之间")
10
        if self.email and "@" not in self.email:
11
            errors.append("email 格式不正确")
12
        return errors
13

14
# 创建实例
15
user = User(name="张三", age=25, email="zhangsan@example.com")
16
print(user)
17
# User(name='张三', age=25, email='zhangsan@example.com')
18

19
# JSON 序列化
20
print(user.to_json())
21
# {
22
#   "name": "张三",
23
#   "age": 25,
24
#   "email": "zhangsan@example.com"
25
# }
26

27
# JSON 反序列化
28
user2 = User.from_json('{"name": "李四", "age": 30}')
29
print(user2)
30
# User(name='李四', age=30, email='')
31

32
# 类型校验
33
try:
34
    bad_user = User(name="王五", age="不是数字")
35
except TypeError as e:
36
    print(f"类型错误: {e}")
37
# 类型错误: age 期望 int, 得到 str
38

39
# 验证
40
invalid_user = User(name="赵六", age=-1, email="invalid")
41
print("验证错误:", invalid_user.validate())
42
# 验证错误: ['age 必须在 0-150 之间', 'email 格式不正确']

1
from functools import wraps
2

3
def validator(field_name: str):
4
    """字段级验证器装饰器"""
5
    def decorator(func):
6
        @wraps(func)
7
        def wrapper(self, value):
8
            result = func(self, value)
9
            if result is not None:
10
                return result
11
            return []
12
        wrapper._validator_field = field_name
13
        return wrapper
14
    return decorator
15

16
class ValidatedModel(Model):
17
    """支持装饰器验证器的模型"""
18
    __validators__: ClassVar[list] = []
19

20
    def __init_subclass__(cls, **kwargs):
21
        super().__init_subclass__(**kwargs)
22
        # 收集所有带 _validator_field 属性的方法
23
        cls.__validators__ = []
24
        for name, method in vars(cls).items():
25
            if callable(method) and hasattr(method, "_validator_field"):
26
                cls.__validators__.append(method)
27

28
    def validate(self) -> list[str]:
29
        errors = super().validate()
30
        for validator_fn in self.__validators__:
31
            field = validator_fn._validator_field
32
            value = getattr(self, field, None)
33
            result = validator_fn(self, value)
34
            if result:
35
                errors.extend(result)
36
        return errors
37

38
class Product(ValidatedModel):
39
    name = Field(str, required=True)
40
    price = Field(float, required=True)
41
    stock = Field(int, required=True, default=0)
42

43
    @validator("price")
44
    def _check_positive_price(self, value):
45
        if value is not None and value <= 0:
46
            return [f"price 必须大于 0，当前: {value}"]
47
        return []
48

49
    @validator("stock")
50
    def _check_non_negative_stock(self, value):
51
        if value is not None and value < 0:
52
            return [f"stock 不能为负数，当前: {value}"]
53
        return []
54

55
product = Product(name="Python 编程", price=79.9, stock=100)
56
print(product.validate())  # []
57

58
bad_product = Product(name="错误商品", price=-10, stock=-5)
59
print(bad_product.validate())
60
# ['price 必须大于 0，当前: -10', 'stock 不能为负数，当前: -5']