1. 动态属性访问:__getattr__、__getattribute__、__setattr__
Python 的属性访问机制是元编程的基石。与普通语言不同,Python 中的每一次属性查找都可以被自定义拦截和重定向。理解这三个魔术方法的调用顺序和差异,是掌握动态特性的第一步。它们在访问控制、属性代理、延迟加载等场景中发挥着至关重要的作用。
1.1 三者的调用顺序与区别
当一个对象执行 obj.attr 时,Python 会按照以下顺序尝试解析属性:
第一步,Python 首先调用 type(obj).__getattribute__(obj, 'attr')。这个方法默认会依次查找数据描述符、实例字典、非数据描述符和类字典。如果 __getattribute__ 本身被重写,则完全由自定义逻辑接管。
第二步,如果 __getattribute__ 在查找过程中抛出 AttributeError,则触发 __getattr__ 作为最终兜底。这使得开发者可以优雅地处理缺失属性,例如实现默认值、动态属性生成或属性代理。
class Demo:
x = 10
def __getattribute__(self, name):
print(f"__getattribute__ 拦截: {name}")
return super().__getattribute__(name)
def __getattr__(self, name):
print(f"__getattr__ 兜底: {name}")
return f"默认值: {name}"
def __setattr__(self, name, value):
print(f"__setattr__ 拦截: {name} = {value}")
super().__setattr__(name, value)
d = Demo()
print(d.x) # __getattribute__ 拦截 → 返回 10
print(d.y) # __getattribute__ 拦截 → AttributeError → __getattr__ 兜底
关键区别:
__getattribute__:拦截所有属性访问,包括实例属性和类属性。重写时需要极其小心,因为任何内部属性访问都会触发递归。__getattr__:仅当正常查找失败(抛出AttributeError)时被调用。它是实现属性代理和默认值的理想场所。__setattr__:拦截所有属性赋值操作,是实现类型校验、不可变性和审计日志的关键点。
注意:
__getattribute__中若直接访问self.xxx会触发无限递归,必须通过super().__getattribute__(name)绕过。这是初学者最容易踩的坑。
1.2 实现不可变配置对象
利用 __setattr__ 和 __getattr__ 的组合,可以构建初始化后完全只读的字典包装器,适合存储应用程序配置等不应被意外修改的数据。
class FrozenDict:
"""初始化后不可修改的字典包装器"""
def __init__(self, data):
# 绕过 __setattr__ 拦截,避免递归
super().__setattr__('_data', dict(data))
def __getattr__(self, name):
try:
return self._data[name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
if hasattr(self, '_data'):
raise AttributeError("FrozenDict is read-only")
super().__setattr__(name, value)
def __delattr__(self, name):
raise AttributeError("FrozenDict is read-only")
def __repr__(self):
return f"FrozenDict({self._data!r})"
config = FrozenDict({'host': 'localhost', 'port': 3306, 'debug': False})
print(config.host) # localhost
# config.port = 5432 # AttributeError: FrozenDict is read-only
1.3 属性代理模式
__getattr__ 常被用于实现代理模式,将属性访问透明地转发到另一个对象上去。例如 Django ORM 的 RelatedManager 就是典型的代理应用。
class LazyLoader:
"""延迟加载代理:仅在首次访问时触发加载逻辑"""
def __init__(self, factory):
self._factory = factory
self._loaded = False
self._target = None
def __getattr__(self, name):
if not self._loaded:
print(f"首次访问 '{name}',触发加载...")
self._target = self._factory()
self._loaded = True
return getattr(self._target, name)
# 模拟一个昂贵的初始化操作
def load_heavy_data():
return {"users": 10000, "orders": 500000, "products": 5000}
stats = LazyLoader(load_heavy_data)
print(stats.users) # 首次访问 'users',触发加载... → 10000
print(stats.orders) # 已加载,直接返回 → 500000
2. 描述符协议(Descriptor Protocol)
描述符是 Python 属性访问的底层机制,也是整个语言面向对象系统中最精妙的设计之一。property、classmethod、staticmethod 本质上都是描述符。掌握描述符协议后,你就可以从底层实现诸如 ORM 字段映射、类型校验、属性缓存等高级功能。
2.1 描述符的核心协议
如果一个类定义了 __get__、__set__ 或 __delete__ 中的任意一个方法,该类的实例就成为一个描述符。当这个实例被作为另一个类的类属性使用时,属性访问会被描述符协议接管。
class Descriptor:
def __init__(self, name):
self.name = name
def __get__(self, instance, owner):
if instance is None:
# 通过类访问时返回描述符自身,支持类级别的操作
return self
print(f"__get__: instance={instance}, owner={owner}")
return instance.__dict__.get(self.name)
def __set__(self, instance, value):
print(f"__set__: instance={instance}, value={value}")
instance.__dict__[self.name] = value
def __delete__(self, instance):
print(f"__delete__: instance={instance}")
del instance.__dict__[self.name]
class Person:
name = Descriptor('name')
age = Descriptor('age')
p = Person()
p.name = "Alice" # __set__ 拦截
print(p.name) # __get__ 拦截 → Alice
数据描述符与非数据描述符:同时定义了 __get__ 和 __set__ 的描述符称为数据描述符(如 property);只定义了 __get__ 的称为非数据描述符(如 classmethod、staticmethod)。两者在属性查找优先级上存在微妙差异,数据描述符拥有更高的查找优先级。
2.2 用描述符实现类型校验字段
这是描述符最常见的生产应用场景之一。Django ORM 的 CharField、IntegerField 等字段本质上就是这一模式的工业级实现。
from typing import Any
class TypedField:
"""带类型校验的描述符字段"""
def __init__(self, name: str, expected_type: type):
self.name = name
self.expected_type = expected_type
def __get__(self, instance: Any, owner: type) -> Any:
if instance is None:
return self
return instance.__dict__.get(self.name)
def __set__(self, instance: Any, value: Any) -> None:
if not isinstance(value, self.expected_type):
raise TypeError(
f"字段 '{self.name}' 期望类型 {self.expected_type.__name__}, "
f"实际传入 {type(value).__name__}"
)
instance.__dict__[self.name] = value
def __delete__(self, instance: Any) -> None:
raise AttributeError(f"字段 '{self.name}' 不可删除")
class Product:
name = TypedField('name', str)
price = TypedField('price', (int, float))
stock = TypedField('stock', int)
product = Product()
product.name = "iPhone 15"
product.price = 6999.00
product.stock = 100
# product.price = "expensive" # TypeError: 字段 'price' 期望类型 float, 实际传入 str
2.3 property 的本质
很多人在使用 @property 时并不了解其底层实现。实际上,property 就是一个内置的数据描述符类。下面的代码展示了其基本实现原理。
# property 等价于以下自定义描述符实现
class MyProperty:
def __init__(self, fget=None, fset=None, fdel=None, doc=None):
self.fget = fget
self.fset = fset
self.fdel = fdel
if doc is None and fget is not None:
doc = fget.__doc__
self.__doc__ = doc
def __get__(self, instance, owner):
if instance is None:
return self
if self.fget is None:
raise AttributeError("unreadable attribute")
return self.fget(instance)
def __set__(self, instance, value):
if self.fset is None:
raise AttributeError("can't set attribute")
self.fset(instance, value)
def setter(self, fset):
return type(self)(self.fget, fset, self.fdel, self.__doc__)
class Circle:
def __init__(self, radius):
self._radius = radius
@MyProperty
def area(self):
"""圆的面积"""
return 3.14159 * self._radius ** 2
@MyProperty
def radius(self):
return self._radius
@radius.setter
def radius(self, value):
if value < 0:
raise ValueError("半径不能为负数")
self._radius = value
c = Circle(5)
print(c.area) # 78.53975
c.radius = 10
print(c.area) # 314.159
3. 元类(Metaclass)
元类是创建类的类,是 Python 面向对象系统中最强大的特性之一。如果说类是对象的模板,那么元类就是类的模板。默认情况下,所有类都由内置的 type 创建。通过自定义元类,你几乎可以控制类创建和初始化的每一个步骤。
3.1 type() 动态创建类
type(name, bases, namespace) 是元类的最朴素用法,也是理解类动态创建机制的起点。这在需要根据配置或运行时参数生成类的框架中非常常见。
# 语法: type(name, bases, namespace)
User = type('User', (), {
'__init__': lambda self, name: setattr(self, 'name', name),
'greet': lambda self: f"Hello, {self.name}!",
})
u = User("Alice")
print(u.greet()) # Hello, Alice!
3.2 自定义元类
最常见的元类应用场景是实现单例模式。相比于在类中使用 __new__,元类实现更加清晰和通用,因为它是将控制逻辑上移到了类的创建层面。
class SingletonMeta(type):
"""单例元类:确保每个类只有一个实例"""
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class Database(metaclass=SingletonMeta):
def __init__(self, dsn: str):
self.dsn = dsn
print(f"初始化数据库连接: {dsn}")
class Logger(metaclass=SingletonMeta):
def __init__(self, level: str):
self.level = level
print(f"初始化日志器: {level}")
db1 = Database("postgresql://localhost/db")
db2 = Database("mysql://localhost/db")
assert db1 is db2 # 同一个实例
3.3 __new__ 与 __init__ 的精确时机
在元类中,__new__ 负责创建类对象本身(此时类尚未存在),而 __init__ 负责对已创建的类对象进行初始化。两者分别对应类和实例生命周期中的 “构造” 与 “初始化” 阶段。
class AutoRegisterMeta(type):
"""自动注册子类的元类"""
registry = {}
def __new__(mcs, name, bases, namespace, **kwargs):
print(f"__new__: 创建类对象 {name}")
# 在类对象创建前可以修改命名空间
if 'version' not in namespace:
namespace['version'] = 1.0
# 创建类对象(此时类尚未存在)
cls = super().__new__(mcs, name, bases, namespace)
return cls
def __init__(cls, name, bases, namespace, **kwargs):
print(f"__init__: 初始化类 {name}")
super().__init__(name, bases, namespace)
# 将类注册到全局表
if name != 'BasePlugin':
AutoRegisterMeta.registry[name] = cls
class BasePlugin(metaclass=AutoRegisterMeta):
pass
class EmailPlugin(BasePlugin):
pass
class SmsPlugin(BasePlugin):
pass
print(AutoRegisterMeta.registry)
# {'EmailPlugin': <class '__main__.EmailPlugin'>, 'SmsPlugin': ...}
3.4 元类冲突与解决
当多重继承的父类使用不同的元类时,Python 无法确定应该使用哪个元类来创建子类,从而抛出 metaclass conflict 错误。解决方案是创建一个公共的派生元类。
class MetaA(type):
pass
class MetaB(type):
pass
class A(metaclass=MetaA):
pass
class B(metaclass=MetaB):
pass
# class C(A, B): pass # TypeError: metaclass conflict
# 解决方案:创建公共派生元类
class MetaAB(MetaA, MetaB):
pass
class C(A, B, metaclass=MetaAB):
pass
print(type(C)) # <class '__main__.MetaAB'>
4. 类装饰器 vs 元类
类装饰器和元类都能修改或增强类的行为,但它们介入的时机和语义有本质区别。类装饰器作用于已经创建好的类对象上,而元类则在类创建的过程中介入。在大多数日常场景中,类装饰器更加直观和易用;而在框架级别的约定或需要深入干预类构造过程时,元类更加不可替代。
# ---------- 类装饰器 ----------
def add_method(cls):
"""为类动态添加方法和属性"""
cls.created_at = "2024-01-01"
def to_dict(self):
return {k: v for k, v in self.__dict__.items() if not k.startswith('_')}
cls.to_dict = to_dict
return cls
@add_method
class User:
def __init__(self, name, email):
self.name = name
self.email = email
u = User("Bob", "bob@example.com")
print(u.to_dict()) # {'name': 'Bob', 'email': 'bob@example.com'}
# ---------- 元类 ----------
class AutoReprMeta(type):
"""自动生成 __repr__ 的元类"""
def __new__(mcs, name, bases, namespace):
if '__repr__' not in namespace:
def auto_repr(self):
attrs = ", ".join(f"{k}={v!r}" for k, v in self.__dict__.items())
return f"{name}({attrs})"
namespace['__repr__'] = auto_repr
return super().__new__(mcs, name, bases, namespace)
class Point(metaclass=AutoReprMeta):
def __init__(self, x, y):
self.x = x
self.y = y
p = Point(1, 2)
print(p) # Point(x=1, y=2)
选择建议:
| 场景 | 推荐方案 | 理由 |
|---|---|---|
| 简单增强(添加方法/属性/统计信息) | 类装饰器 | 更直观、可组合、易单元测试 |
| 控制类创建过程或修改属性定义 | 元类 | 可以修改 __new__ 行为,在类创建前介入 |
| 需要与继承体系深度集成 | 元类 | 子类自动继承元类行为,无需显式装饰 |
| 框架级别的透明约定(ORM、API 路由) | 元类 | 对开发者完全透明,只需正常定义类即可 |
5. __slots__ 与内存优化
默认情况下,Python 实例使用 __dict__ 存储属性。这是一个动态哈希表,虽然灵活但内存开销较大。对于需要创建大量实例的场景(如游戏对象、大规模数据处理),__slots__ 可以显著降低内存占用并提升属性访问速度。
class RegularUser:
"""普通类:每个实例都有独立的 __dict__ 字典"""
def __init__(self, name, age, email):
self.name = name
self.age = age
self.email = email
class SlottedUser:
"""使用 __slots__:固定属性集合,禁止动态添加新属性"""
__slots__ = ['name', 'age', 'email']
def __init__(self, name, age, email):
self.name = name
self.age = age
self.email = email
# 内存对比
import sys
r = RegularUser("Alice", 30, "alice@example.com")
s = SlottedUser("Alice", 30, "alice@example.com")
print(sys.getsizeof(r)) # 约 152 bytes
print(sys.getsizeof(s)) # 约 64 bytes(显著降低,约为前者的 42%)
# 额外特性差异
r.phone = "123456" # 普通类可以动态添加属性
# s.phone = "123456" # SlottedUser 会抛出 AttributeError
注意事项:
__slots__中若未显式包含'__dict__',实例将没有__dict__,这意味着无法动态添加新属性。如果需要保留动态属性能力,可以写成__slots__ = ['name', '__dict__']。- 子类不会继承父类的
__slots__定义。即使父类使用了__slots__,子类若未定义自己的__slots__,实例仍然会持有__dict__。 - 使用
__slots__会提升属性访问速度,因为属性存储从字典查找变为了基于偏移量的数组访问。 __slots__的限制还包括不能有__weakref__(除非显式声明),且多继承时的注意事项更多。
6. Monkey Patching 及其风险
Monkey Patching 是指在运行时动态修改模块、类或对象的行为。这个名字来源于将程序的各个部分随意地拼接在一起的过程。在某些场景下(如测试环境模拟、热修复第三方库 Bug),Monkey Patching 提供了极大的便利,但同时也带来了严峻的可维护性挑战。
import datetime
# 场景:测试环境统一冻结时间,使日期相关的测试具有确定性
original_now = datetime.datetime.now
class FreezableDatetime(type):
_frozen = None
@classmethod
def freeze(cls, year, month, day):
cls._frozen = datetime.datetime(year, month, day)
@classmethod
def now(cls):
if cls._frozen:
return cls._frozen
return original_now()
# Monkey Patch:替换标准库行为
datetime.datetime.now = FreezableDatetime.now
# 测试使用
FreezableDatetime.freeze(2024, 1, 15)
print(datetime.datetime.now()) # 2024-01-15 00:00:00
6.1 主要风险分析
- 维护困难:运行时修改使得代码行为难以预测和追踪。审查代码时无法仅靠静态分析判断实际调用的方法。
- 协作冲突:多个补丁可能互相覆盖,导致诡异且难以复现的 Bug。尤其是不同依赖库之间独立进行补丁时。
- 版本脆弱:被补丁的对象在依赖库升级后,内部实现可能改变,补丁可能失效或产生错误行为。
- 测试污染:全局修改可能影响其他测试用例的执行环境,导致测试之间的隐性依赖和污染。
6.2 安全实践
from contextlib import contextmanager
@contextmanager
def patch_method(target, method_name, replacement):
"""上下文安全的 Monkey Patch:退出上下文后自动恢复原方法"""
original = getattr(target, method_name)
setattr(target, method_name, replacement)
try:
yield
finally:
setattr(target, method_name, original)
# 使用示例
with patch_method(SomeService, 'send', mock_send):
run_tests()
# 退出 with 语句块后,SomeService.send 已自动恢复原实现
7. 内省工具(Introspection)
内省是指程序在运行时检查自身结构和状态的能力。Python 提供了丰富的内省工具,使代码能够动态感知函数签名、类结构、调用栈等信息。这是构建调试器、IDE 插件、代码生成器和自动化测试框架的基础能力。
7.1 inspect 模块
inspect 是 Python 标准库中最强大的内省工具,涵盖了从函数签名到调用栈的各个方面。
import inspect
def sample(a: int, b: str = "default", *args, **kwargs) -> bool:
"""示例函数:展示 inspect 的各种用法"""
return True
# 获取函数签名
sig = inspect.signature(sample)
for name, param in sig.parameters.items():
print(f"{name}: kind={param.kind.name}, default={param.default}, annotation={param.annotation}")
# 检查对象类型
print(inspect.isfunction(sample)) # True
print(inspect.isgeneratorfunction(sample)) # False
# 获取源码(要求对象为 .py 文件中的定义,不适用于交互式会话)
print(inspect.getsource(sample))
# 获取调用帧(实现高级调试工具、日志追踪)
def who_called_me():
frame = inspect.currentframe().f_back
print(f"被 {frame.f_code.co_name} 调用,文件: {frame.f_code.co_filename}:{frame.f_lineno}")
7.2 getattr、hasattr、setattr 与 vars
这些内置函数是内省最基础也是最常用的工具,配合得当可以实现极其灵活的配置解析和插件加载逻辑。
class Config:
DEBUG = False
DATABASE_URL = "postgresql://localhost/db"
TIMEOUT = 30
# 动态访问:根据环境变量决定读取哪个配置项
env = "production"
key = "DEBUG" if env == "development" else "DATABASE_URL"
print(getattr(Config, key, "fallback"))
# 动态检查并条件调用
if hasattr(Config, 'setup'):
getattr(Config, 'setup')()
# vars() 等价于对象的 __dict__
obj = type('Obj', (), {'a': 1, 'b': 2})()
print(vars(obj)) # {'a': 1, 'b': 2}
7.3 __dir__ 与自定义内省
覆盖 __dir__ 方法可以控制 dir() 的输出结果,在实现包装器或代理对象时特别有用。
class FilteredObject:
def __init__(self):
self.public = "可见"
self._private = "隐藏"
self._internal = "隐藏"
def __dir__(self):
return [k for k in self.__dict__ if not k.startswith('_')]
obj = FilteredObject()
print(dir(obj)) # ['public'] ,_private 和 _internal 被过滤
8. 代码生成:exec / eval 与安全考量
exec 和 eval 是 Python 中最直接的动态代码执行机制,能够将字符串形式的代码转换为实际的运行指令。这种能力虽然强大,但同时也引入了严重的安全风险,因为恶意输入可能导致任意代码执行。
8.1 基本用法
# eval: 求值表达式(返回结果)
expr = "2 ** 10 + 100"
result = eval(expr)
print(result) # 1124
# exec: 执行语句(无返回值)
code = """
def dynamic_greet(name):
return f"Hello, {name}!"
result = dynamic_greet("World")
"""
namespace = {}
exec(code, namespace)
print(namespace['result']) # Hello, World!
8.2 安全的受限执行
在生产环境中,如果必须使用 eval,必须严格限制可访问的名称空间,彻底移除内置函数引用。
def safe_eval(expr: str, allowed_names: dict = None):
"""受限表达式求值:仅允许白名单中的名称,无任何内置函数访问权"""
allowed_names = allowed_names or {}
code = compile(expr, "<string>", "eval")
# 检查代码中引用的所有全局名称是否均在白名单中
for name in code.co_names:
if name not in allowed_names:
raise NameError(f"'{name}' 不在允许的名称列表中")
return eval(code, {"__builtins__": {}}, allowed_names)
# 使用
variables = {"x": 10, "y": 20}
print(safe_eval("(x + y) * 2", variables)) # 60
# safe_eval("__import__('os').system('rm -rf /')") # NameError
8.3 动态生成类
import types
def create_model(name: str, fields: dict):
"""根据字段定义动态生成数据模型类"""
annotations = {}
init_body = []
for field_name, field_type in fields.items():
annotations[field_name] = field_type
init_body.append(f" self.{field_name} = {field_name}")
init_source = "def __init__(self, " + ", ".join(fields) + "):\n"
init_source += "\n".join(init_body)
namespace = {'__annotations__': annotations}
exec(init_source, namespace)
return type(name, (), {
'__init__': namespace['__init__'],
'__annotations__': annotations,
'__repr__': lambda self: (
f"{name}(" + ", ".join(
f"{k}={getattr(self, k)!r}" for k in fields
) + ")"
),
})
PersonModel = create_model("Person", {"name": str, "age": int})
p = PersonModel("Alice", 30)
print(p) # Person(name='Alice', age=30)
生产环境中应优先使用
ast模块解析代码结构,或使用types.new_class替代exec,以从根本上杜绝代码注入风险。
9. 实战案例
9.1 ORM 字段描述符
结合描述符、元类和类型注解,可以实现一个极简但功能完备的 ORM 模型框架。这是理解 Django ORM 和 SQLAlchemy 底层原理的最佳切入点。
from typing import Any, Optional
class Field:
"""ORM 字段基类"""
def __init__(self, name: str, field_type: str, primary_key: bool = False,
nullable: bool = True, default: Any = None):
self.name = name
self.field_type = field_type
self.primary_key = primary_key
self.nullable = nullable
self.default = default
def __get__(self, instance: Any, owner: type) -> Any:
if instance is None:
return self
return instance.__dict__.get(self.name, self.default)
def __set__(self, instance: Any, value: Any) -> None:
if value is None and not self.nullable:
raise ValueError(f"字段 '{self.name}' 不可为空")
instance.__dict__[self.name] = value
def __repr__(self) -> str:
return f"Field({self.name}: {self.field_type})"
class StringField(Field):
def __init__(self, name: str, max_length: int = 255, **kwargs):
super().__init__(name, f"VARCHAR({max_length})", **kwargs)
self.max_length = max_length
def __set__(self, instance: Any, value: Any) -> None:
if value is not None and len(str(value)) > self.max_length:
raise ValueError(f"'{self.name}' 超过最大长度 {self.max_length}")
super().__set__(instance, value)
class IntegerField(Field):
def __init__(self, name: str, **kwargs):
super().__init__(name, "INTEGER", **kwargs)
class ModelMeta(type):
"""ORM 模型元类:自动收集 Field 描述符并注册到类属性中"""
def __new__(mcs, name: str, bases: tuple, namespace: dict):
cls = super().__new__(mcs, name, bases, namespace)
cls._fields = {}
for attr_name, attr_value in namespace.items():
if isinstance(attr_value, Field):
attr_value.name = attr_name # 自动绑定字段名
cls._fields[attr_name] = attr_value
return cls
class Model(metaclass=ModelMeta):
"""ORM 模型基类"""
def __init__(self, **kwargs):
for field_name, field in self._fields.items():
value = kwargs.get(field_name, field.default)
setattr(self, field_name, value)
def to_dict(self) -> dict:
return {name: getattr(self, name) for name in self._fields}
@classmethod
def create_table_sql(cls) -> str:
columns = []
for name, field in cls._fields.items():
constraints = []
if field.primary_key:
constraints.append("PRIMARY KEY")
if not field.nullable:
constraints.append("NOT NULL")
col = f" {name} {field.field_type} {' '.join(constraints)}"
columns.append(col)
return f"CREATE TABLE {cls.__name__.lower()} (\n" + ",\n".join(columns) + "\n);"
class User(Model):
id = IntegerField('id', primary_key=True, nullable=False)
username = StringField('username', max_length=50, nullable=False)
email = StringField('email', max_length=120, nullable=False)
age = IntegerField('age', default=0)
user = User(id=1, username="alice", email="alice@example.com")
print(user.to_dict())
print(User.create_table_sql())
9.2 API 路由自动注册
利用类方法装饰器实现一个简洁的 API 路由注册系统,类似 Flask 和 FastAPI 的装饰器路由机制。
from functools import wraps
class APIRegistry:
"""API 路由注册中心"""
_routes = {}
@classmethod
def register(cls, path: str, methods: tuple = ("GET",)):
def decorator(func):
cls._routes[path] = {
'handler': func,
'methods': methods,
'doc': func.__doc__,
}
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
@classmethod
def dispatch(cls, path: str, method: str = "GET"):
route = cls._routes.get(path)
if not route or method not in route['methods']:
raise ValueError(f"路由 {path} [{method}] 未找到")
return route['handler']
@APIRegistry.register("/users", methods=("GET", "POST"))
def list_users():
"""获取用户列表"""
return {"users": ["alice", "bob"]}
@APIRegistry.register("/users/<id>", methods=("GET", "PUT", "DELETE"))
def user_detail(user_id: int):
"""获取/更新/删除指定用户"""
return {"user_id": user_id}
# 查看注册的路由
for path, info in APIRegistry._routes.items():
print(f"{path}: {info['methods']} - {info['doc']}")
9.3 插件系统
设计一个基于元类和抽象基类的插件系统,实现插件的自动发现、注册和条件执行。这种架构在数据处理流水线、命令行工具扩展和 ETL 框架中非常常见。
import importlib
from abc import ABC, abstractmethod
class PluginMeta(type):
"""插件元类:继承自 BasePlugin 的类自动注册到全局注册表"""
registry = {}
def __init__(cls, name, bases, namespace):
super().__init__(name, bases, namespace)
if name != 'BasePlugin' and hasattr(cls, 'plugin_name'):
PluginMeta.registry[cls.plugin_name] = cls
class BasePlugin(ABC, metaclass=PluginMeta):
"""插件基类:所有插件必须实现 process 和 validate 方法"""
plugin_name = ""
@abstractmethod
def process(self, data: dict) -> dict:
pass
@abstractmethod
def validate(self, data: dict) -> bool:
pass
class JsonFormatterPlugin(BasePlugin):
plugin_name = "json_formatter"
def process(self, data: dict) -> dict:
return {"formatted": str(data), "type": "json"}
def validate(self, data: dict) -> bool:
return isinstance(data, dict)
class UpperCasePlugin(BasePlugin):
plugin_name = "upper_case"
def process(self, data: dict) -> dict:
return {k: str(v).upper() for k, v in data.items()}
def validate(self, data: dict) -> bool:
return all(isinstance(v, str) for v in data.values())
class PluginManager:
"""插件管理器:负责动态加载插件并执行数据处理流水线"""
def __init__(self):
self._active_plugins = []
def load_plugin(self, name: str):
if name not in PluginMeta.registry:
raise ValueError(f"插件 '{name}' 未注册")
self._active_plugins.append(PluginMeta.registry[name]())
def load_from_module(self, module_name: str):
"""动态导入模块并触发插件注册"""
importlib.import_module(module_name)
def execute(self, data: dict) -> list:
results = []
for plugin in self._active_plugins:
if plugin.validate(data):
results.append(plugin.process(data))
else:
results.append({"error": f"{type(plugin).__name__} 验证失败"})
return results
# 使用
manager = PluginManager()
manager.load_plugin("json_formatter")
manager.load_plugin("upper_case")
result = manager.execute({"name": "alice", "role": "admin"})
for r in result:
print(r)
总结
Python 的元编程能力赋予了开发者极高的灵活性和抽象能力,各个核心机制之间形成了一个层次分明的技术栈:
- 描述符控制单个属性的访问行为,是
property和 ORM 字段绑定的底层支撑。掌握了描述符,你就掌握了 Python 属性系统的核心。 - 动态属性访问(
__getattr__/__getattribute__)处理属性查找的 fallback 逻辑,是实现代理、延迟加载和配置对象的利器。 - 元类控制类的创建过程,适合框架级别的约定(ORM 模型注册、自动代码生成、类行为增强)。它是修改类行为最彻底的方式。
- 类装饰器以非侵入方式增强类,适合简单的横切关注点。相比元类,它的学习成本更低、可组合性更强。
__slots__在性能敏感场景下替代繁重的__dict__,可以将实例内存占用降低超过一半,同时获得更快的属性访问速度。inspect和反射工具使运行时代码能够 “观察” 自身的结构,是实现调试器、IDE、自动化测试框架和 API 文档生成器的基础。exec/eval虽然强大,但应严格限制输入范围或完全避免在安全敏感的场景中使用,ast模块通常是更安全的选择。
合理运用这些动态特性,可以在不牺牲代码可读性的前提下,构建出高度抽象、可扩展、低耦合的框架与工具链。元编程能力的真正价值不在于炫技,而在于将复杂重复的模式内化为简洁优雅的代码约定。
延伸阅读
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。