第31章 Python元编程模式
学习目标
完成本章学习后，读者将能够：
- 理解Python元编程的核心概念与形式化定义
- 掌握装饰器、描述符、元类等高级技术的数学原理
- 实现属性验证、惰性加载、自动注册等元编程模式
- 应用元编程技术构建领域特定语言(DSL)
31.1 理论基础与形式化定义
31.1.1 元编程的形式化定义
定义 31.1（元编程）：元编程是一种编程范式，程序可以将其他程序（或自身）作为数据进行操作。形式化表示为：
$$MetaProgram: Program \rightarrow Program$$
定义 31.2（装饰器）：装饰器是一个高阶函数，接受一个函数并返回一个增强的函数：
$$decorator: (A \rightarrow B) \rightarrow (A \rightarrow B)$$
定义 31.3（描述符协议）：描述符是一个实现以下方法之一的对象：
$$Descriptor = \langle __get__, __set__, __delete__ \rangle$$
其中：
- $get(self, obj, type) \rightarrow value$
- $set(self, obj, value) \rightarrow void$
- $delete(self, obj) \rightarrow void$
定义 31.4（元类）：元类是类的类，定义类的创建行为：
$$Metaclass: (name, bases, namespace) \rightarrow Class$$
31.1.2 Python元编程层次
┌─────────────────────────────────────────────────────────────────────────┐
│ Python元编程层次模型 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 元类层 (Metaclass) │ │
│ │ type │ 自定义元类 │ __new__ │ __init_subclass__ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 类装饰器层 (Class Decorator) │ │
│ │ @classmethod │ @staticmethod │ 自定义类装饰器 │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 描述符层 (Descriptor) │ │
│ │ property │ 自定义描述符 │ __get__ │ __set__ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 装饰器层 (Decorator) │ │
│ │ 函数装饰器 │ 带参数装饰器 │ 类装饰器 │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 动态特性层 (Dynamic Features) │ │
│ │ __getattr__ │ __setattr__ │ __getattribute__ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
31.1.3 历史背景与发展脉络
| 时期 | Python版本 | 特性 | 核心贡献 |
|---|---|---|---|
| 1991 | Python 1.0 | 基础反射 | getattr, setattr |
| 2000 | Python 2.0 | 列表推导 | 函数式特性 |
| 2001 | Python 2.2 | 描述符协议 | property, 统一类型 |
| 2003 | Python 2.3 | 装饰器语法 | @decorator |
| 2006 | Python 2.6 | 抽象基类 | ABCMeta |
| 2012 | Python 3.0 | 元类语法 | metaclass参数 |
| 2015 | Python 3.5 | 类型注解 | typing模块 |
| 2018 | Python 3.7 | dataclass | 自动生成方法 |
| 2020 | Python 3.9 | 类型提示增强 | 泛型语法简化 |
| 2021+ | Python 3.10+ | 模式匹配 | 结构化模式匹配 |
31.2 装饰器模式
31.2.1 装饰器的形式化定义
定义 31.5（装饰器组合）：多个装饰器的组合满足结合律：
$$(d_1 \circ d_2) \circ d_3 = d_1 \circ (d_2 \circ d_3)$$
装饰器的应用顺序为从下到上、从内到外。
31.2.2 UML类图
┌─────────────────────────────────────────────────────────────────────────┐
│ <<interface>> │
│ Callable[P, T] │
├─────────────────────────────────────────────────────────────────────────┤
│ + __call__(*args: P.args, **kwargs: P.kwargs): T │
└─────────────────────────────────────────────────────────────────────────┘
△
│
┌───────────────────────────┼───────────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Function │ │ Decorator │ │ ClassDecorator│
├───────────────┤ ├───────────────┤ ├───────────────┤
│ - func │ │ - wrapper │ │ - cls │
├───────────────┤ ├───────────────┤ ├───────────────┤
│ + __call__() │ │ + __call__() │ │ + __call__() │
└───────────────┘ └───────────────┘ └───────────────┘
┌─────────────────────────────────────────────────────────────────────────┐
│ <<decorator>> │
│ ParametrizedDecorator │
├─────────────────────────────────────────────────────────────────────────┤
│ - params: Dict │
├─────────────────────────────────────────────────────────────────────────┤
│ + __call__(func: Callable) -> Callable │
└─────────────────────────────────────────────────────────────────────────┘1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
31.2.3 函数装饰器完整实现
python
from typing import Callable, Any, Dict, List, Optional, TypeVar, ParamSpec
from functools import wraps
import time
import inspect
from dataclasses import dataclass
import warnings
P = ParamSpec('P')
T = TypeVar('T')
def timer(func: Callable[P, T]) -> Callable[P, T]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} 执行时间: {elapsed:.6f}秒")
return result
return wrapper
def retry(
max_attempts: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
exceptions: tuple = (Exception,)
) -> Callable[[Callable[P, T]], Callable[P, T]]:
def decorator(func: Callable[P, T]) -> Callable[P, T]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
last_exception = None
current_delay = delay
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
time.sleep(current_delay)
current_delay *= backoff
raise last_exception
return wrapper
return decorator
def validate(*validators: Callable[[Any], bool]) -> Callable:
def decorator(func: Callable) -> Callable:
sig = inspect.signature(func)
@wraps(func)
def wrapper(*args, **kwargs):
bound = sig.bind(*args, **kwargs)
bound.apply_defaults()
for name, value in bound.arguments.items():
for validator in validators:
if not validator(value):
raise ValueError(f"参数 {name}={value} 验证失败")
return func(*args, **kwargs)
return wrapper
return decorator
def memoize(func: Callable[P, T]) -> Callable[P, T]:
cache: Dict[tuple, T] = {}
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
key = (args, frozenset(kwargs.items()))
if key not in cache:
cache[key] = func(*args, **kwargs)
return cache[key]
wrapper.cache = cache
wrapper.cache_clear = lambda: cache.clear()
wrapper.cache_info = lambda: {"size": len(cache), "keys": list(cache.keys())}
return wrapper
def deprecated(
message: str = None,
version: str = None,
alternative: str = None
) -> Callable:
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
parts = [message or f"函数 {func.__name__} 已废弃"]
if version:
parts.append(f"自版本 {version} 起")
if alternative:
parts.append(f"请使用 {alternative} 替代")
warnings.warn(" ".join(parts), DeprecationWarning, stacklevel=2)
return func(*args, **kwargs)
return wrapper
return decorator
def log_calls(
level: str = "INFO",
include_args: bool = True,
include_result: bool = True
) -> Callable:
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
func_name = func.__qualname__
if include_args:
args_repr = [repr(a) for a in args]
kwargs_repr = [f"{k}={repr(v)}" for k, v in kwargs.items()]
signature = ", ".join(args_repr + kwargs_repr)
print(f"[{level}] 调用 {func_name}({signature})")
else:
print(f"[{level}] 调用 {func_name}")
try:
result = func(*args, **kwargs)
if include_result:
print(f"[{level}] {func_name} 返回: {repr(result)}")
return result
except Exception as e:
print(f"[ERROR] {func_name} 抛出异常: {type(e).__name__}: {e}")
raise
return wrapper
return decorator
def trace(depth: int = 2) -> Callable:
import traceback
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
stack = traceback.extract_stack()[-depth-1:-1]
caller = stack[0] if stack else None
if caller:
print(f"[TRACE] {func.__name__} 被调用于 {caller.filename}:{caller.lineno}")
return func(*args, **kwargs)
return wrapper
return decorator
@timer
@retry(max_attempts=3, delay=0.5)
@memoize
def fibonacci(n: int) -> int:
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
@log_calls(level="DEBUG")
def process_data(data: List[int], multiplier: int = 2) -> List[int]:
return [x * multiplier for x in data]
print(f"fibonacci(10) = {fibonacci(10)}")
print(f"缓存信息: {fibonacci.cache_info()}")
result = process_data([1, 2, 3], multiplier=3)
print(f"结果: {result}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
31.2.4 类装饰器实现
python
from typing import Any, Dict, List, Callable, Type, TypeVar
from functools import wraps
from dataclasses import dataclass
import json
import threading
T = TypeVar('T')
def singleton(cls: Type[T]) -> Type[T]:
instances: Dict[Type, Any] = {}
lock = threading.Lock()
@wraps(cls)
def wrapper(*args, **kwargs) -> T:
if cls not in instances:
with lock:
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
wrapper._instance = None
wrapper._instances = instances
return wrapper
def add_logging(cls: Type[T]) -> Type[T]:
for name, method in cls.__dict__.items():
if callable(method) and not name.startswith('_'):
setattr(cls, name, log_calls()(method))
return cls
def immutable(cls: Type[T]) -> Type[T]:
original_init = cls.__init__
original_setattr = cls.__setattr__ if hasattr(cls, '__setattr__') else object.__setattr__
@wraps(original_init)
def new_init(self, *args, **kwargs):
object.__setattr__(self, '_initialized', False)
original_init(self, *args, **kwargs)
object.__setattr__(self, '_initialized', True)
def new_setattr(self, name, value):
if getattr(self, '_initialized', False):
raise AttributeError(f"不可变类 {cls.__name__} 不允许修改属性 {name}")
original_setattr(self, name, value)
cls.__init__ = new_init
cls.__setattr__ = new_setattr
return cls
def serializable(cls: Type[T]) -> Type[T]:
def to_dict(self) -> Dict:
result = {}
for key, value in self.__dict__.items():
if not key.startswith('_'):
if hasattr(value, 'to_dict'):
result[key] = value.to_dict()
else:
result[key] = value
return result
def to_json(self) -> str:
return json.dumps(self.to_dict(), ensure_ascii=False)
@classmethod
def from_dict(cls: Type[T], data: Dict) -> T:
return cls(**data)
@classmethod
def from_json(cls: Type[T], json_str: str) -> T:
return cls.from_dict(json.loads(json_str))
cls.to_dict = to_dict
cls.to_json = to_json
cls.from_dict = from_dict
cls.from_json = from_json
return cls
def validate_types(cls: Type[T]) -> Type[T]:
original_init = cls.__init__
annotations = getattr(cls, '__annotations__', {})
@wraps(original_init)
def new_init(self, *args, **kwargs):
original_init(self, *args, **kwargs)
for attr_name, expected_type in annotations.items():
if hasattr(self, attr_name):
actual_value = getattr(self, attr_name)
if not isinstance(actual_value, expected_type):
raise TypeError(
f"属性 {attr_name} 应为 {expected_type.__name__}，"
f"实际为 {type(actual_value).__name__}"
)
cls.__init__ = new_init
return cls
def cached_property(func: Callable) -> property:
attr_name = func.__name__
@wraps(func)
def getter(self):
if attr_name not in self.__dict__:
self.__dict__[attr_name] = func(self)
return self.__dict__[attr_name]
return property(getter)
@singleton
class Database:
def __init__(self, connection_string: str):
self.connection_string = connection_string
print(f"创建数据库连接: {connection_string}")
db1 = Database("mysql://localhost")
db2 = Database("mysql://localhost")
print(f"单例验证: {db1 is db2}")
@serializable
@validate_types
class User:
name: str
age: int
email: str
def __init__(self, name: str, age: int, email: str):
self.name = name
self.age = age
self.email = email
user = User("张三", 25, "zhangsan@example.com")
print(f"JSON: {user.to_json()}")
user2 = User.from_json('{"name": "李四", "age": 30, "email": "lisi@example.com"}')
print(f"反序列化: {user2.name}, {user2.age}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
31.2.5 带状态的装饰器
python
from typing import Callable, Any, Optional, Dict
from functools import wraps
import time
class CountCalls:
def __init__(self, func: Callable):
self.func = func
self.count = 0
wraps(func)(self)
def __call__(self, *args, **kwargs) -> Any:
self.count += 1
print(f"调用 {self.func.__name__} 第 {self.count} 次")
return self.func(*args, **kwargs)
def reset(self) -> None:
self.count = 0
def get_count(self) -> int:
return self.count
class RateLimit:
def __init__(self, calls_per_second: float = 1.0):
self.calls_per_second = calls_per_second
self.min_interval = 1.0 / calls_per_second
self.last_call = 0.0
def __call__(self, func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - self.last_call
if elapsed < self.min_interval:
wait_time = self.min_interval - elapsed
print(f"限流: 等待 {wait_time:.2f}秒")
time.sleep(wait_time)
self.last_call = time.time()
return func(*args, **kwargs)
return wrapper
class CacheWithExpiry:
def __init__(self, ttl_seconds: float = 60.0, max_size: int = 100):
self.ttl = ttl_seconds
self.max_size = max_size
self.cache: Dict[str, tuple] = {}
def __call__(self, func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
key = str((args, frozenset(kwargs.items())))
current_time = time.time()
if key in self.cache:
value, timestamp = self.cache[key]
if current_time - timestamp < self.ttl:
print(f"缓存命中: {func.__name__}")
return value
if len(self.cache) >= self.max_size:
oldest_key = min(self.cache.keys(), key=lambda k: self.cache[k][1])
del self.cache[oldest_key]
result = func(*args, **kwargs)
self.cache[key] = (result, current_time)
return result
wrapper.cache = self.cache
wrapper.clear_cache = lambda: self.cache.clear()
wrapper.cache_info = lambda: {
"size": len(self.cache),
"max_size": self.max_size,
"ttl": self.ttl
}
return wrapper
class Throttle:
def __init__(self, period: float = 1.0):
self.period = period
self.last_called = 0.0
def __call__(self, func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
if now - self.last_called >= self.period:
self.last_called = now
return func(*args, **kwargs)
else:
print(f"节流: 跳过 {func.__name__} 调用")
return wrapper
class Debounce:
def __init__(self, wait: float = 0.1):
self.wait = wait
self._timer = None
self._last_args = None
self._last_kwargs = None
def __call__(self, func: Callable) -> Callable:
import threading
@wraps(func)
def wrapper(*args, **kwargs):
self._last_args = args
self._last_kwargs = kwargs
if self._timer:
self._timer.cancel()
def execute():
func(*self._last_args, **self._last_kwargs)
self._timer = threading.Timer(self.wait, execute)
self._timer.start()
return wrapper
@CountCalls
def greet(name: str) -> str:
return f"Hello, {name}!"
print(greet("Alice"))
print(greet("Bob"))
print(f"总调用次数: {greet.get_count()}")
@CacheWithExpiry(ttl_seconds=5.0, max_size=10)
def expensive_computation(n: int) -> int:
print(f"计算 fibonacci({n})")
if n < 2:
return n
return expensive_computation(n-1) + expensive_computation(n-2)
print(expensive_computation(10))
print(f"缓存信息: {expensive_computation.cache_info()}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
31.3 描述符模式
31.3.1 描述符的形式化定义
定义 31.6（描述符协议）：描述符是一个对象，实现以下方法：
$$Descriptor = {d \mid d.__get__ \lor d.__set__ \lor d.__delete__ }$$
定义 31.7（数据描述符）：实现 $set$ 或 $delete$ 的描述符：
$$DataDescriptor = {d \mid d.__set__ \in d \lor d.__delete__ \in d}$$
定义 31.8（非数据描述符）：仅实现 $get$ 的描述符：
$$NonDataDescriptor = {d \mid d.__get__ \in d \land d.__set__ \notin d}$$
31.3.2 UML类图
┌─────────────────────────────────────────────────────────────────────────┐
│ <<abstract>> │
│ Descriptor │
├─────────────────────────────────────────────────────────────────────────┤
│ - name: str │
├─────────────────────────────────────────────────────────────────────────┤
│ + __set_name__(owner, name): void │
│ + __get__(obj, owner): Any │
│ + __set__(obj, value): void │
│ + __delete__(obj): void │
└─────────────────────────────────────────────────────────────────────────┘
△
│
┌───────────────────────────┼───────────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Typed │ │ Validated │ │ Lazy │
├───────────────┤ ├───────────────┤ ├───────────────┤
│ - ty: Type │ │ - validators │ │ - func │
├───────────────┤ ├───────────────┤ ├───────────────┤
│ + validate() │ │ + validate() │ │ + __get__() │
└───────────────┘ └───────────────┘ └───────────────┘
│ │
▼ ▼
┌───────────────┐ ┌───────────────┐
│ Integer │ │ Email │
│ Float │ │ Phone │
│ String │ │ Range │
└───────────────┘ └───────────────┘1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31.3.3 基础描述符实现
python
from typing import Any, Optional, Type, TypeVar, Generic, Callable, List
from weakref import WeakKeyDictionary
from dataclasses import dataclass
import re
T = TypeVar('T')
class Descriptor:
def __init__(self, name: str = None, default: Any = None):
self.name = name
self.default = default
def __set_name__(self, owner: Type, name: str) -> None:
self.name = name
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
return obj.__dict__.get(self.name, self.default)
def __set__(self, obj: Any, value: Any) -> None:
obj.__dict__[self.name] = value
def __delete__(self, obj: Any) -> None:
if self.name in obj.__dict__:
del obj.__dict__[self.name]
class Typed(Descriptor):
ty: Type = object
def __set__(self, obj: Any, value: Any) -> None:
if value is not None and not isinstance(value, self.ty):
raise TypeError(
f"属性 {self.name} 期望类型 {self.ty.__name__}，"
f"实际类型 {type(value).__name__}"
)
super().__set__(obj, value)
class Integer(Typed):
ty = int
class Float(Typed):
ty = float
class String(Typed):
ty = str
class Boolean(Typed):
ty = bool
class Positive(Descriptor):
def __set__(self, obj: Any, value: Any) -> None:
if value is not None and value < 0:
raise ValueError(f"属性 {self.name} 必须为正数，得到 {value}")
super().__set__(obj, value)
class PositiveInteger(Integer, Positive):
pass
class PositiveFloat(Float, Positive):
pass
class Sized(Descriptor):
def __init__(
self,
*args,
min_size: int = None,
max_size: int = None,
**kwargs
):
self.min_size = min_size
self.max_size = max_size
super().__init__(*args, **kwargs)
def __set__(self, obj: Any, value: Any) -> None:
if value is not None:
length = len(value)
if self.min_size is not None and length < self.min_size:
raise ValueError(
f"属性 {self.name} 长度不能小于 {self.min_size}，"
f"实际长度 {length}"
)
if self.max_size is not None and length > self.max_size:
raise ValueError(
f"属性 {self.name} 长度不能大于 {self.max_size}，"
f"实际长度 {length}"
)
super().__set__(obj, value)
class SizedString(String, Sized):
pass
class Range(Descriptor):
def __init__(
self,
*args,
min_val: Any = None,
max_val: Any = None,
**kwargs
):
self.min_val = min_val
self.max_val = max_val
super().__init__(*args, **kwargs)
def __set__(self, obj: Any, value: Any) -> None:
if value is not None:
if self.min_val is not None and value < self.min_val:
raise ValueError(
f"属性 {self.name} 不能小于 {self.min_val}，"
f"实际值 {value}"
)
if self.max_val is not None and value > self.max_val:
raise ValueError(
f"属性 {self.name} 不能大于 {self.max_val}，"
f"实际值 {value}"
)
super().__set__(obj, value)
class BoundedInteger(Integer, Range):
pass
class BoundedFloat(Float, Range):
pass
class Choice(Descriptor):
def __init__(self, *args, choices: List[Any] = None, **kwargs):
self.choices = choices or []
super().__init__(*args, **kwargs)
def __set__(self, obj: Any, value: Any) -> None:
if value is not None and value not in self.choices:
raise ValueError(
f"属性 {self.name} 必须是 {self.choices} 之一，"
f"实际值 {value}"
)
super().__set__(obj, value)
class Person:
name = SizedString(min_size=2, max_size=50)
age = PositiveInteger()
salary = PositiveFloat()
def __init__(self, name: str, age: int, salary: float):
self.name = name
self.age = age
self.salary = salary
person = Person("张三", 25, 5000.0)
print(f"姓名: {person.name}, 年龄: {person.age}, 薪资: {person.salary}")
try:
person.age = -5
except ValueError as e:
print(f"验证错误: {e}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
31.3.4 属性验证描述符
python
from typing import Callable, List, Any, Optional, Pattern
from dataclasses import dataclass
import re
@dataclass
class ValidationError:
field: str
message: str
value: Any
class Validated(Descriptor):
validators: List[Callable[[str, Any], None]] = []
def __set__(self, obj: Any, value: Any) -> None:
self.validate(value)
super().__set__(obj, value)
def validate(self, value: Any) -> None:
for validator in self.validators:
validator(self.name, value)
class Email(String):
EMAIL_PATTERN = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
def __set__(self, obj: Any, value: Any) -> None:
if value is not None:
if not self.EMAIL_PATTERN.match(value):
raise ValueError(f"属性 {self.name} 不是有效的邮箱地址: {value}")
super().__set__(obj, value)
class Phone(String):
PHONE_PATTERN = re.compile(r'^1[3-9]\d{9}$')
def __set__(self, obj: Any, value: Any) -> None:
if value is not None:
if not self.PHONE_PATTERN.match(value):
raise ValueError(f"属性 {self.name} 不是有效的手机号码: {value}")
super().__set__(obj, value)
class URL(String):
URL_PATTERN = re.compile(
r'^https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+[/?:%.\w]*$'
)
def __set__(self, obj: Any, value: Any) -> None:
if value is not None:
if not self.URL_PATTERN.match(value):
raise ValueError(f"属性 {self.name} 不是有效的URL: {value}")
super().__set__(obj, value)
class Regex(String):
def __init__(self, *args, pattern: str = None, **kwargs):
self.pattern = re.compile(pattern) if pattern else None
super().__init__(*args, **kwargs)
def __set__(self, obj: Any, value: Any) -> None:
if value is not None and self.pattern:
if not self.pattern.match(value):
raise ValueError(
f"属性 {self.name} 不匹配模式 {self.pattern.pattern}: {value}"
)
super().__set__(obj, value)
class CustomValidator(Descriptor):
def __init__(self, *args, validator: Callable[[Any], bool] = None, **kwargs):
self.validator = validator
super().__init__(*args, **kwargs)
def __set__(self, obj: Any, value: Any) -> None:
if value is not None and self.validator:
if not self.validator(value):
raise ValueError(f"属性 {self.name} 验证失败: {value}")
super().__set__(obj, value)
class Customer:
name = SizedString(min_size=2, max_size=100)
email = Email()
phone = Phone()
age = BoundedInteger(min_val=0, max_val=150)
level = Choice(choices=["普通", "银卡", "金卡", "钻石"])
website = URL(required=False)
def __init__(
self,
name: str,
email: str,
phone: str,
age: int,
level: str,
website: str = None
):
self.name = name
self.email = email
self.phone = phone
self.age = age
self.level = level
if website:
self.website = website
customer = Customer(
name="李四",
email="lisi@example.com",
phone="13800138000",
age=30,
level="金卡"
)
print(f"客户: {customer.name}, 等级: {customer.level}")
try:
customer.email = "invalid-email"
except ValueError as e:
print(f"验证错误: {e}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
31.3.5 惰性属性描述符
python
from typing import Callable, Any, Type, Dict
from functools import wraps
from weakref import WeakKeyDictionary
class LazyProperty:
def __init__(self, func: Callable):
self.func = func
self.attr_name = func.__name__
wraps(func)(self)
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
if self.attr_name not in obj.__dict__:
print(f"惰性计算属性: {self.attr_name}")
obj.__dict__[self.attr_name] = self.func(obj)
return obj.__dict__[self.attr_name]
class CachedProperty:
def __init__(self, func: Callable):
self.func = func
self.attr_name = func.__name__
wraps(func)(self)
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
cache_key = f"_cached_{self.attr_name}"
if not hasattr(obj, cache_key):
print(f"缓存计算属性: {self.attr_name}")
setattr(obj, cache_key, self.func(obj))
return getattr(obj, cache_key)
def __delete__(self, obj: Any) -> None:
cache_key = f"_cached_{self.attr_name}"
if hasattr(obj, cache_key):
delattr(obj, cache_key)
class ObservableProperty:
def __init__(self, default: Any = None):
self.default = default
self.data = WeakKeyDictionary()
self.callbacks: Dict[int, List[Callable]] = {}
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
return self.data.get(obj, self.default)
def __set__(self, obj: Any, value: Any) -> None:
old_value = self.data.get(obj, self.default)
self.data[obj] = value
obj_id = id(obj)
if obj_id in self.callbacks:
for callback in self.callbacks[obj_id]:
callback(old_value, value)
def observe(self, obj: Any, callback: Callable) -> None:
obj_id = id(obj)
if obj_id not in self.callbacks:
self.callbacks[obj_id] = []
self.callbacks[obj_id].append(callback)
class ComputedProperty:
def __init__(self, compute_func: Callable, dependencies: List[str] = None):
self.compute_func = compute_func
self.dependencies = dependencies or []
self._cache: Dict[int, tuple] = {}
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
obj_id = id(obj)
dep_values = tuple(
getattr(obj, dep, None) for dep in self.dependencies
)
if obj_id in self._cache:
cached_deps, cached_value = self._cache[obj_id]
if cached_deps == dep_values:
return cached_value
value = self.compute_func(obj)
self._cache[obj_id] = (dep_values, value)
return value
class DataProcessor:
@LazyProperty
def expensive_data(self) -> list:
print("执行耗时计算...")
return [i ** 2 for i in range(100000)]
@CachedProperty
def config(self) -> dict:
print("加载配置...")
return {"timeout": 30, "retries": 3}
status = ObservableProperty(default="idle")
processor = DataProcessor()
print("访问惰性属性:")
print(f"数据长度: {len(processor.expensive_data)}")
print(f"再次访问: {len(processor.expensive_data)}")
print("\n访问缓存属性:")
print(f"配置: {processor.config}")
print(f"再次访问: {processor.config}")
def on_status_change(old: str, new: str):
print(f"状态变更: {old} -> {new}")
DataProcessor.status.observe(processor, on_status_change)
processor.status = "running"
processor.status = "completed"1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
31.4 元类模式
31.4.1 元类的形式化定义
定义 31.9（元类）：元类是创建类的类。在Python中，所有类都是type的实例：
$$\forall C: Class \Rightarrow isinstance(C, type)$$
定义 31.10（类创建过程）：类创建遵循以下步骤：
$$create_class = __call__ \rightarrow __new__ \rightarrow __init__$$
31.4.2 UML类图
┌─────────────────────────────────────────────────────────────────────────┐
│ type │
├─────────────────────────────────────────────────────────────────────────┤
│ + __new__(mcs, name, bases, namespace): Type │
│ + __init__(cls, name, bases, namespace): void │
│ + __call__(cls, *args, **kwargs): Instance │
└─────────────────────────────────────────────────────────────────────────┘
△
│
│ extends
│
┌─────────────────────────────────────────────────────────────────────────┐
│ <<abstract>> │
│ CustomMeta │
├─────────────────────────────────────────────────────────────────────────┤
│ - _registry: Dict[str, Type] │
├─────────────────────────────────────────────────────────────────────────┤
│ + __new__(mcs, name, bases, namespace): Type │
│ + __call__(cls, *args, **kwargs): Instance │
└─────────────────────────────────────────────────────────────────────────┘
△
│
┌───────────────────────────┼───────────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ SingletonMeta │ │ RegistryMeta │ │ InterfaceMeta │
├───────────────┤ ├───────────────┤ ├───────────────┤
│ - _instances │ │ - _registry │ │ + validate() │
├───────────────┤ ├───────────────┤ └───────────────┘
│ + __call__() │ │ + get_plugin()│
└───────────────┘ └───────────────┘1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
31.4.3 基础元类实现
python
from typing import Any, Dict, List, Type, Tuple, Optional
from abc import ABCMeta
import time
import threading
class Meta(type):
def __new__(mcs, name: str, bases: Tuple[Type, ...],
namespace: Dict[str, Any]) -> Type:
print(f"[Meta.__new__] 创建类: {name}")
namespace['_created_at'] = time.time()
namespace['_class_name'] = name
cls = super().__new__(mcs, name, bases, namespace)
return cls
def __init__(cls, name: str, bases: Tuple[Type, ...],
namespace: Dict[str, Any]) -> None:
print(f"[Meta.__init__] 初始化类: {name}")
super().__init__(name, bases, namespace)
def __call__(cls, *args, **kwargs) -> Any:
print(f"[Meta.__call__] 创建实例: {cls.__name__}")
instance = super().__call__(*args, **kwargs)
return instance
class SingletonMeta(type):
_instances: Dict[Type, Any] = {}
_lock = threading.Lock()
def __call__(cls, *args, **kwargs) -> Any:
if cls not in cls._instances:
with cls._lock:
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class RegistryMeta(type):
_registry: Dict[str, Type] = {}
def __new__(mcs, name: str, bases: Tuple[Type, ...],
namespace: Dict[str, Any]) -> Type:
cls = super().__new__(mcs, name, bases, namespace)
if name != 'BasePlugin':
mcs._registry[name] = cls
return cls
@classmethod
def get_plugin(mcs, name: str) -> Optional[Type]:
return mcs._registry.get(name)
@classmethod
def list_plugins(mcs) -> List[str]:
return list(mcs._registry.keys())
@classmethod
def create_instance(mcs, name: str, *args, **kwargs) -> Optional[Any]:
plugin_class = mcs.get_plugin(name)
if plugin_class:
return plugin_class(*args, **kwargs)
return None
class Document(metaclass=Meta):
def __init__(self, title: str):
self.title = title
doc = Document("测试文档")
class Cache(metaclass=SingletonMeta):
def __init__(self):
self._data: Dict[str, Any] = {}
def get(self, key: str) -> Any:
return self._data.get(key)
def set(self, key: str, value: Any) -> None:
self._data[key] = value
cache1 = Cache()
cache2 = Cache()
print(f"单例验证: {cache1 is cache2}")
class BasePlugin(metaclass=RegistryMeta):
def execute(self) -> str:
raise NotImplementedError
class EmailPlugin(BasePlugin):
def execute(self) -> str:
return "发送邮件"
class SMSPlugin(BasePlugin):
def execute(self) -> str:
return "发送短信"
print(f"已注册插件: {RegistryMeta.list_plugins()}")
print(f"创建实例: {RegistryMeta.create_instance('EmailPlugin').execute()}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
31.4.4 自动注册元类
python
from typing import Dict, Type, Any, Optional, List, Tuple
from enum import Enum
class AutoRegisterMeta(type):
_registry: Dict[str, Type] = {}
_by_type: Dict[Any, Type] = {}
def __new__(mcs, name: str, bases: Tuple[Type, ...],
namespace: Dict[str, Any]) -> Type:
cls = super().__new__(mcs, name, bases, namespace)
if hasattr(cls, '_register_key'):
key = cls._register_key
mcs._registry[name] = cls
mcs._by_type[key] = cls
return cls
@classmethod
def get_by_name(mcs, name: str) -> Optional[Type]:
return mcs._registry.get(name)
@classmethod
def get_by_type(mcs, type_key: Any) -> Optional[Type]:
return mcs._by_type.get(type_key)
@classmethod
def all_registered(mcs) -> Dict[str, Type]:
return mcs._registry.copy()
@classmethod
def all_types(mcs) -> Dict[Any, Type]:
return mcs._by_type.copy()
class MessageType(Enum):
EMAIL = "email"
SMS = "sms"
PUSH = "push"
WEBHOOK = "webhook"
class MessageHandler(metaclass=AutoRegisterMeta):
_register_key: MessageType = None
def handle(self, message: dict) -> dict:
raise NotImplementedError
class EmailHandler(MessageHandler):
_register_key = MessageType.EMAIL
def handle(self, message: dict) -> dict:
return {
"status": "sent",
"channel": "email",
"recipient": message.get("email")
}
class SMSHandler(MessageHandler):
_register_key = MessageType.SMS
def handle(self, message: dict) -> dict:
return {
"status": "sent",
"channel": "sms",
"recipient": message.get("phone")
}
class PushHandler(MessageHandler):
_register_key = MessageType.PUSH
def handle(self, message: dict) -> dict:
return {
"status": "sent",
"channel": "push",
"device": message.get("device_id")
}
def send_message(msg_type: MessageType, message: dict) -> dict:
handler_class = AutoRegisterMeta.get_by_type(msg_type)
if handler_class:
handler = handler_class()
return handler.handle(message)
return {"status": "error", "message": "未知的消息类型"}
print(f"已注册处理器: {list(AutoRegisterMeta.all_registered().keys())}")
result = send_message(MessageType.EMAIL, {"email": "test@example.com"})
print(f"发送结果: {result}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
31.4.5 接口检查元类
python
from typing import Set, Dict, Any, Tuple, Type, List
from abc import ABCMeta, abstractmethod
class InterfaceMeta(ABCMeta):
def __new__(mcs, name: str, bases: Tuple[Type, ...],
namespace: Dict[str, Any]) -> Type:
cls = super().__new__(mcs, name, bases, namespace)
if not namespace.get('_is_interface'):
mcs._validate_implementation(cls)
return cls
@staticmethod
def _validate_implementation(cls: Type) -> None:
required_methods: Set[str] = set()
for base in cls.__mro__[1:]:
if hasattr(base, '_is_interface') and base._is_interface:
for attr_name in dir(base):
attr = getattr(base, attr_name)
if getattr(attr, '__isabstractmethod__', False):
required_methods.add(attr_name)
missing = []
for method in required_methods:
if method not in cls.__dict__:
missing.append(method)
if missing:
raise TypeError(
f"类 {cls.__name__} 未实现接口方法: {', '.join(missing)}"
)
class RepositoryInterface(metaclass=InterfaceMeta):
_is_interface = True
@abstractmethod
def find_by_id(self, id: Any) -> Any:
pass
@abstractmethod
def save(self, entity: Any) -> None:
pass
@abstractmethod
def delete(self, id: Any) -> None:
pass
@abstractmethod
def find_all(self) -> List[Any]:
pass
class UserRepository(RepositoryInterface):
def __init__(self, db_connection: str):
self.db = db_connection
self._users: Dict[int, dict] = {}
def find_by_id(self, id: int) -> dict:
return self._users.get(id)
def save(self, entity: dict) -> None:
self._users[entity['id']] = entity
def delete(self, id: int) -> None:
if id in self._users:
del self._users[id]
def find_all(self) -> List[dict]:
return list(self._users.values())
user_repo = UserRepository("mysql://localhost")
user_repo.save({"id": 1, "name": "张三"})
print(f"用户: {user_repo.find_by_id(1)}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
31.5 动态属性模式
31.5.1 动态属性的形式化定义
定义 31.11（属性访问协议）：属性访问遵循以下查找顺序：
$$getattr(obj, name) = \begin{cases} obj.__dict__[name] & \text{if } name \in obj.__dict__ \ type(obj).__dict__[name].__get__(obj) & \text{if data descriptor} \ obj.__dict__[name] & \text{if } name \in obj.__dict__ \ type(obj).__dict__[name].__get__(obj) & \text{if non-data descriptor} \ type(obj).__dict__[name] & \text{if } name \in type(obj).__dict__ \ obj.__getattr__(name) & \text{otherwise} \end{cases}$$
31.5.2 动态属性实现
python
from typing import Any, Dict, Callable, Optional
from dataclasses import dataclass
class DynamicAttributes:
def __init__(self, **kwargs):
self._data: Dict[str, Any] = {}
self._setters: Dict[str, Callable] = {}
self._getters: Dict[str, Callable] = {}
for key, value in kwargs.items():
setattr(self, key, value)
def __getattr__(self, name: str) -> Any:
if name.startswith('_'):
raise AttributeError(f"'{type(self).__name__}' 对象没有属性 '{name}'")
if name in self._getters:
return self._getters[name](self._data.get(name))
if name in self._data:
return self._data[name]
raise AttributeError(f"'{type(self).__name__}' 对象没有属性 '{name}'")
def __setattr__(self, name: str, value: Any) -> None:
if name.startswith('_'):
super().__setattr__(name, value)
return
if name in self._setters:
value = self._setters[name](value)
self._data[name] = value
def __delattr__(self, name: str) -> None:
if name in self._data:
del self._data[name]
else:
super().__delattr__(name)
def register_setter(self, name: str, setter: Callable) -> None:
self._setters[name] = setter
def register_getter(self, name: str, getter: Callable) -> None:
self._getters[name] = getter
def to_dict(self) -> Dict[str, Any]:
return self._data.copy()
class ChainAccess:
def __init__(self):
self._data: Dict[str, Any] = {}
def __getattr__(self, name: str) -> 'ChainAccess':
if name.startswith('_'):
raise AttributeError(f"无效属性: {name}")
if name not in self._data:
self._data[name] = ChainAccess()
return self._data[name]
def __call__(self, *args, **kwargs) -> Any:
if args:
return args[0]
return kwargs
def set(self, value: Any) -> 'ChainAccess':
self._value = value
return self
def get(self, default: Any = None) -> Any:
return getattr(self, '_value', default)
class LazyLoader:
def __init__(self, loader: Callable[[], Any]):
self._loader = loader
self._loaded = False
self._value = None
def __getattr__(self, name: str) -> Any:
if not self._loaded:
self._value = self._loader()
self._loaded = True
return getattr(self._value, name)
def __call__(self, *args, **kwargs) -> Any:
if not self._loaded:
self._value = self._loader()
self._loaded = True
return self._value(*args, **kwargs)
obj = DynamicAttributes(name="张三", age=25)
obj.register_setter("age", lambda x: max(0, min(150, x)))
obj.age = 200
print(f"年龄(限制后): {obj.age}")
chain = ChainAccess()
chain.user.profile.name.set("李四")
print(f"链式访问: {chain.user.profile.name.get()}")
def load_heavy_data():
print("加载重量级数据...")
return {"key": "value", "count": 100}
lazy = LazyLoader(load_heavy_data)
print("惰性加载器已创建")
print(f"访问数据: {lazy.key}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
31.5.3 属性代理模式
python
from typing import Any, Dict, Type, Optional, Callable, List
from weakref import WeakKeyDictionary
class PropertyProxy:
def __init__(self, target_getter: Callable, target_setter: Callable = None):
self._target_getter = target_getter
self._target_setter = target_setter
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
return self._target_getter(obj)
def __set__(self, obj: Any, value: Any) -> None:
if self._target_setter is None:
raise AttributeError("属性只读")
self._target_setter(obj, value)
class DeepProperty:
def __init__(self, path: str):
self.path = path.split('.')
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
current = obj
for attr in self.path:
if hasattr(current, attr):
current = getattr(current, attr)
elif isinstance(current, dict):
current = current.get(attr)
else:
return None
return current
def __set__(self, obj: Any, value: Any) -> None:
current = obj
for attr in self.path[:-1]:
if hasattr(current, attr):
current = getattr(current, attr)
elif isinstance(current, dict):
current = current.setdefault(attr, {})
final_attr = self.path[-1]
if isinstance(current, dict):
current[final_attr] = value
else:
setattr(current, final_attr, value)
class AliasedProperty:
def __init__(self, source_attr: str):
self.source_attr = source_attr
def __get__(self, obj: Any, owner: Type = None) -> Any:
if obj is None:
return self
return getattr(obj, self.source_attr)
def __set__(self, obj: Any, value: Any) -> None:
setattr(obj, self.source_attr, value)
class Config:
def __init__(self):
self._settings = {
"database": {
"host": "localhost",
"port": 5432,
"credentials": {
"username": "admin",
"password": "secret"
}
},
"cache": {
"enabled": True,
"ttl": 300
}
}
db_host = DeepProperty("_settings.database.host")
db_port = DeepProperty("_settings.database.port")
db_username = DeepProperty("_settings.database.credentials.username")
cache_enabled = DeepProperty("_settings.cache.enabled")
cache_ttl = DeepProperty("_settings.cache.ttl")
config = Config()
print(f"数据库主机: {config.db_host}")
print(f"数据库端口: {config.db_port}")
config.db_host = "192.168.1.100"
print(f"新主机: {config.db_host}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
31.6 反模式与最佳实践
31.6.1 常见反模式
| 反模式 | 描述 | 后果 | 解决方案 |
|---|---|---|---|
| 过度装饰 | 嵌套过多装饰器 | 代码难以理解 | 合并相关装饰器 |
| 元类滥用 | 简单问题用元类解决 | 增加复杂度 | 优先使用装饰器 |
| 描述符状态共享 | 描述符实例状态共享 | 数据污染 | 使用WeakKeyDictionary |
| 无限递归 | __getattr__调用自身 | 栈溢出 | 正确处理属性访问 |
| 忽略functools.wraps | 装饰器不保留元信息 | 调试困难 | 使用@wraps装饰器 |
31.6.2 最佳实践清单
python
class MetaprogrammingBestPractices:
DECORATOR_RULES = [
"使用functools.wraps保留元信息",
"保持装饰器简单，复杂逻辑封装在单独函数",
"带参数的装饰器返回装饰器函数",
"类装饰器实现__call__方法",
"使用类型注解提高可读性"
]
DESCRIPTOR_RULES = [
"使用__set_name__获取属性名",
"使用WeakKeyDictionary避免内存泄漏",
"数据描述符优先级高于实例字典",
"非数据描述符优先级低于实例字典",
"为描述符添加清晰的文档"
]
METACLASS_RULES = [
"优先使用__init_subclass__替代元类",
"元类继承关系需要保持一致",
"避免多重继承中的元类冲突",
"元类只在类创建时执行一次",
"为元类添加清晰的文档"
]
PERFORMANCE_RULES = [
"装饰器开销很小，可以放心使用",
"描述符访问比直接属性访问稍慢",
"__getattr__只在属性不存在时调用",
"缓存计算结果避免重复计算",
"避免在热路径中使用元编程"
]1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
31.7 决策指南
31.7.1 元编程技术选择
┌─────────────────────────────────────┐
│ 需要修改什么？ │
└─────────────────────────────────────┘
│
┌───────────────┴───────────────┐
│ │
▼ ▼
┌───────────────┐ ┌───────────────┐
│ 函数行为 │ │ 类行为 │
└───────────────┘ └───────────────┘
│ │
▼ ▼
┌───────────────┐ ┌───────────────┐
│ 装饰器 │ │ 是否需要控制创建？│
└───────────────┘ └───────────────┘
│
┌───────────────┴───────────────┐
│ │
▼ ▼
┌───────────────┐ ┌───────────────┐
│ 元类 │ │ 类装饰器 │
└───────────────┘ └───────────────┘1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
31.7.2 技术选择对照表
| 场景 | 推荐技术 | 说明 |
|---|---|---|
| 函数增强 | 装饰器 | 日志、缓存、验证 |
| 属性控制 | 描述符 | 类型验证、惰性加载 |
| 类定制 | 元类/类装饰器 | 自动注册、接口检查 |
| 动态属性 | getattr | 代理、动态加载 |
| 简单类定制 | init_subclass | 避免元类复杂度 |
31.8 快速参考卡片
31.8.1 核心概念速查
┌─────────────────────────────────────────────────────────────────────────┐
│ Python元编程核心概念速查表 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ 装饰器模式 │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ @decorator │ │
│ │ def func(): pass │ │
│ │ 等价于: func = decorator(func) │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ 描述符协议 │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ __get__(self, obj, owner) -> value │ │
│ │ __set__(self, obj, value) -> None │ │
│ │ __delete__(self, obj) -> None │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ 元类创建 │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ class Meta(type): │ │
│ │ def __new__(mcs, name, bases, namespace): │ │
│ │ return super().__new__(mcs, name, bases, namespace) │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ 属性查找顺序 │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 1. 数据描述符.__get__ │ │
│ │ 2. obj.__dict__ │ │
│ │ 3. 非数据描述符.__get__ │ │
│ │ 4. type(obj).__dict__ │ │
│ │ 5. obj.__getattr__ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
31.8.2 代码模板速查
python
# 函数装饰器模板
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
# 带参数装饰器模板
def decorator_with_args(arg):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
# 描述符模板
class MyDescriptor:
def __set_name__(self, owner, name):
self.name = name
def __get__(self, obj, owner):
if obj is None:
return self
return obj.__dict__.get(self.name)
def __set__(self, obj, value):
obj.__dict__[self.name] = value
# 元类模板
class MyMeta(type):
def __new__(mcs, name, bases, namespace):
cls = super().__new__(mcs, name, bases, namespace)
return cls1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
31.9 小结
31.9.1 核心要点
- 装饰器：高阶函数，增强函数或类行为
- 描述符：实现属性访问协议，控制属性行为
- 元类：控制类创建过程，实现高级定制
- 动态属性：通过__getattr__实现动态属性访问
- 属性查找顺序：数据描述符 > 实例字典 > 非数据描述符
31.9.2 适用场景
| 适用 | 不适用 |
|---|---|
| 框架开发 | 简单应用 |
| 代码复用 | 性能敏感场景 |
| DSL构建 | 团队不熟悉 |
| 自动化任务 | 需要调试的场景 |
31.9.3 实施建议
- 保持简单：优先选择简单的解决方案
- 文档清晰：为元编程代码添加详细文档
- 测试充分：元编程代码需要更充分的测试
- 性能考虑：避免在热路径中使用复杂元编程
- 团队共识：确保团队成员理解元编程代码