Python 装饰器深度指南：从语法糖到元编程利器

1
from functools import wraps
2
import time
3

4
def cache(ttl=60):
5
    """带 TTL 的简易缓存装饰器"""
6
    def decorator(func):
7
        _cache = {}
8

9
        @wraps(func)
10
        def wrapper(*args, **kwargs):
11
            # 构造缓存键（注意：kwargs 需要排序以保证一致性）
12
            key = (args, tuple(sorted(kwargs.items())))
13

14
            if key in _cache:
15
                result, timestamp = _cache[key]
16
                if time.time() - timestamp < ttl:
17
                    print(f"🎯 缓存命中: {func.__name__}")
18
                    return result
19
                else:
20
                    del _cache[key]  # 过期清除
21

22
            result = func(*args, **kwargs)
23
            _cache[key] = (result, time.time())
24
            return result
25

26
        # 暴露清除缓存的方法
27
        wrapper.clear_cache = lambda: _cache.clear()
28
        return wrapper
29
    return decorator
30

31
@cache(ttl=10)
32
def get_user_profile(user_id):
33
    """模拟数据库查询"""
34
    print(f"🔍 查询数据库: user_id={user_id}")
35
    time.sleep(0.1)  # 模拟延迟
36
    return {"id": user_id, "name": f"User_{user_id}"}
37

38
# 第一次调用：查数据库
39
profile = get_user_profile(42)
40
# 🔍 查询数据库: user_id=42
41

42
# 第二次调用：走缓存
43
profile = get_user_profile(42)
44
# 🎯 缓存命中: get_user_profile

1
from functools import wraps
2

3
def require_role(*roles):
4
    """检查用户角色的装饰器"""
5
    def decorator(func):
6
        @wraps(func)
7
        def wrapper(*args, **kwargs):
8
            # 假设第一个参数是 request 对象
9
            request = kwargs.get('request') or (args[0] if args else None)
10
            if not request or not hasattr(request, 'user'):
11
                raise PermissionError("未登录")
12

13
            user_role = getattr(request.user, 'role', None)
14
            if user_role not in roles:
15
                raise PermissionError(
16
                    f"需要 {'/'.join(roles)} 权限，当前角色: {user_role}"
17
                )
18
            return func(*args, **kwargs)
19
        return wrapper
20
    return decorator
21

22
# 使用示例
23
@require_role("admin", "superadmin")
24
def delete_user(request, user_id):
25
    return f"用户 {user_id} 已删除"
26

27
@require_role("editor", "admin")
28
def publish_article(request, article_id):
29
    return f"文章 {article_id} 已发布"

1
from functools import wraps
2
import time
3
import logging
4

5
logging.basicConfig(level=logging.INFO)
6
logger = logging.getLogger(__name__)
7

8
def monitor(log_args=True, warn_threshold=1.0):
9
    """生产级监控装饰器：记录调用日志 + 慢查询告警"""
10
    def decorator(func):
11
        @wraps(func)
12
        def wrapper(*args, **kwargs):
13
            # 构造日志上下文
14
            func_name = f"{func.__module__}.{func.__qualname__}"
15
            if log_args:
16
                args_repr = [repr(a) for a in args[:3]]  # 最多记录3个位置参数
17
                kwargs_repr = [f"{k}={v!r}" for k, v in list(kwargs.items())[:3]]
18
                signature = ", ".join(args_repr + kwargs_repr)
19
                if len(args) > 3 or len(kwargs) > 3:
20
                    signature += ", ..."
21
            else:
22
                signature = "..."
23

24
            logger.info(f"▶️ 调用 {func_name}({signature})")
25

26
            start = time.perf_counter()
27
            try:
28
                result = func(*args, **kwargs)
29
                elapsed = time.perf_counter() - start
30

31
                if elapsed > warn_threshold:
32
                    logger.warning(
33
                        f"🐢 慢调用 {func_name}: {elapsed:.3f}s "
34
                        f"(阈值: {warn_threshold}s)"
35
                    )
36
                else:
37
                    logger.info(f"✅ {func_name} 完成: {elapsed:.3f}s")
38

39
                return result
40
            except Exception as e:
41
                elapsed = time.perf_counter() - start
42
                logger.error(
43
                    f"❌ {func_name} 异常: {type(e).__name__}: {e} "
44
                    f"(耗时: {elapsed:.3f}s)"
45
                )
46
                raise
47
        return wrapper
48
    return decorator
49

50
@monitor(warn_threshold=0.5)
51
def query_database(sql, params=None):
52
    time.sleep(0.8)  # 模拟慢查询
53
    return [{"id": 1}]
54

55
query_database("SELECT * FROM users WHERE id = %s", params=(42,))
56
# WARNING - 🐢 慢调用 __main__.query_database: 0.801s (阈值: 0.5s)

1
# ❌ 经典闭包陷阱
2
decorators = []
3
for i in range(3):
4
    def make_decorator(func):
5
        @wraps(func)
6
        def wrapper(*args, **kwargs):
7
            print(f"装饰器 {i}")  # i 始终是 2！
8
            return func(*args, **kwargs)
9
        return wrapper
10
    decorators.append(make_decorator)
11

12
# ✅ 正确做法：用默认参数捕获当前值
13
for i in range(3):
14
    def make_decorator(func, _i=i):  # _i=i 在定义时求值
15
        @wraps(func)
16
        def wrapper(*args, **kwargs):
17
            print(f"装饰器 {_i}")
18
            return func(*args, **kwargs)
19
        return wrapper
20
    decorators.append(make_decorator)