1. 为什么你总在装饰器上卡壳?——从“写不出来”到“随手就来”的真实路径

Python装饰器这个概念,几乎每个学过函数式编程的开发者都听过、看过、甚至抄过几遍示例代码。但真正轮到自己写一个带参数的类装饰器,或者想把日志、重试、缓存三个功能组合起来用时,八成会停下来盯着编辑器发呆: @log @retry @cache 这样写对吗?顺序有影响吗?如果我要给 @retry 加个最大重试次数,是写成 @retry(max_attempts=3) 还是 @retry(3) ?更别提调试时发现被装饰函数的 __name__ 变成了 wrapper help() 一看全是 *args, **kwargs ,连参数名都丢了——这时候你不是不会Python,而是没真正理解装饰器的 执行时机 作用域嵌套本质

我带过二十多个Python项目,从爬虫脚手架到金融风控引擎,凡是用到装饰器的地方,90%以上的bug都出在三类典型场景:一是多层装饰器叠加后行为不可预测(比如 @auth @rate_limit @cache 中缓存键没排除认证头);二是类装饰器里 __call__ __init__ 混用导致实例状态错乱;三是用 functools.wraps 时只复制了 __name__ 却漏了 __doc__ __annotations__ ,结果Pydantic模型校验直接失败。这篇文章不讲“什么是装饰器”,也不堆砌 @decorator 语法糖的定义,而是带你回到最原始的函数对象层面,用四步可验证的操作,把装饰器从“魔法黑箱”变成“透明工具箱”。你会看到:一个带参数的装饰器,本质上就是三次函数调用的嵌套; @ 符号只是语法糖,删掉它代码照样跑;而所谓“简化”,不是减少代码行数,而是让每一步的输入输出都清晰可见、可断点、可单元测试。适合刚写完第一个Flask路由的新手,也适合正在重构微服务中间件的老手——只要你曾为装饰器的闭包变量生命周期挠过头,这篇就是为你写的。

2. 装饰器的本质解构:剥开三层洋葱,看清函数对象的流转

2.1 第一层:装饰器不是语法,而是函数调用链

很多人以为 @log 是某种特殊指令,其实它等价于手动调用: func = log(func) 。关键在于, log 必须是一个 接受函数作为参数并返回新函数 的可调用对象。我们先写一个最简版本,不加任何语法糖:

def log(func):
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__} with {args}, {kwargs}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} returned {result}")
        return result
    return wrapper

def add(a, b):
    return a + b

# 手动装饰:这就是@log的真实含义
add = log(add)
add(2, 3)  # 输出:Calling add with (2, 3), {} → add returned 5

这里没有魔法。 log 接收 add 函数对象,返回 wrapper 函数对象,然后把 add 变量重新指向 wrapper wrapper 内部通过闭包保存了原始 func 的引用,所以能调用它。 重点来了 wrapper 的签名是 (*args, **kwargs) ,这意味着它抹去了原始函数的参数信息。如果你用 inspect.signature(add) 查看,得到的是 (*args, **kwargs) 而不是 (a, b) 。这会导致类型检查工具(如mypy)失效,IDE无法提示参数名,Pydantic解析失败——这些都不是装饰器“不好用”,而是你没补全函数元数据。

提示: functools.wraps 不是万能胶,它是 update_wrapper 的快捷方式,本质是把源函数的 __name__ , __doc__ , __module__ , __annotations__ , __dict__ 等属性复制到目标函数。但注意: __annotations__ 复制后,如果源函数注解了 a: int , b: int wrapper __annotations__ 就会包含这些,IDE才能正确提示。很多教程只写 @wraps(func) 却不提 __annotations__ ,这是实操中踩坑最多的地方。

2.2 第二层:带参数的装饰器——其实是“生成装饰器的工厂函数”

当你看到 @retry(max_attempts=3) ,直觉会觉得 retry 接收了 max_attempts 参数。但语法规定:装饰器符号 @ 后面必须跟一个 可调用对象 ,且该对象接收被装饰函数作为唯一参数。所以 retry(max_attempts=3) 本身必须返回一个符合装饰器接口的函数。这意味着 retry 实际上是一个 工厂函数 ,结构必然是三层嵌套:

def retry(max_attempts=3):
    # 第一层:接收装饰器参数,返回真正的装饰器
    def decorator(func):
        # 第二层:接收被装饰函数,返回包装函数
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise e
                    print(f"Attempt {attempt + 1} failed, retrying...")
        return wrapper
    return decorator  # 返回第二层函数,这才是真正的装饰器

# 使用:@retry(max_attempts=3) 等价于 add = retry(max_attempts=3)(add)
@retry(max_attempts=2)
def unstable_add(a, b):
    import random
    if random.random() < 0.7:  # 70%概率失败
        raise ValueError("Network error")
    return a + b

这个三层结构是硬性约束,无法绕过。 retry 先用 max_attempts 初始化状态, decorator func 初始化闭包, wrapper *args, **kwargs 执行逻辑。 为什么必须这样设计? 因为Python解释器在遇到 @retry(max_attempts=2) 时,会先执行 retry(max_attempts=2) 得到 decorator 函数对象,再用这个对象去装饰 unstable_add 。如果 retry 直接返回 wrapper ,那 @retry(max_attempts=2) 就变成了 unstable_add = retry(max_attempts=2)(unstable_add) ,但 retry(max_attempts=2) 返回的是 wrapper ,而 wrapper(unstable_add) 会报错——因为 wrapper 期望的是 *args, **kwargs ,不是函数对象。

2.3 第三层:类装饰器——用 __call__ 替代闭包,状态管理更清晰

函数装饰器依赖闭包保存状态(如 max_attempts ),但闭包变量在多次装饰时容易混淆。比如你写 @retry(3) 装饰函数A,又 @retry(5) 装饰函数B,两个 wrapper 里的 max_attempts 是独立的,没问题;但如果你在 wrapper 里用了 nonlocal 修改外部变量,或者用 global ,就会出问题。类装饰器用实例属性管理状态,逻辑更直观:

class Retry:
    def __init__(self, max_attempts=3):
        self.max_attempts = max_attempts  # 实例属性,每个装饰器独立
    
    def __call__(self, func):
        # __call__ 使实例可调用,等价于函数装饰器的第二层
        def wrapper(*args, **kwargs):
            for attempt in range(self.max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == self.max_attempts - 1:
                        raise e
                    print(f"Attempt {attempt + 1} failed, retrying...")
        # 关键:wraps必须应用在wrapper上,不是self上
        from functools import wraps
        return wraps(func)(wrapper)

# 使用完全一样
@Retry(max_attempts=2)
def unstable_add(a, b):
    ...

类装饰器的优势在于:状态明确( self.max_attempts )、可继承( class BackoffRetry(Retry): )、可调试( print(retry_instance.max_attempts) )。但要注意: __call__ 方法必须返回一个函数,不能直接执行逻辑; wraps 依然要作用于返回的 wrapper ,否则元数据丢失。我见过太多人把 wraps 错放在 __init__ __call__ 上,结果 help() 里还是 (*args, **kwargs)

3. 四步实操法:从零写出可维护、可测试、可组合的装饰器

3.1 第一步:定义接口契约——先写测试,再写实现

不要一上来就敲 def decorator(func): 。先问三个问题:

  1. 这个装饰器要改变什么行为?(如:增加日志、修改返回值、拦截异常)
  2. 它需要哪些配置参数?(如:日志级别、重试间隔、缓存过期时间)
  3. 它是否应该保持原始函数的签名和文档?(99%的情况答案是“是”)

@cache 为例,我们先写单元测试,明确接口:

import pytest
from unittest.mock import patch

def test_cache_decorator():
    # 测试1:首次调用执行原函数,后续调用返回缓存
    call_count = 0
    @cache(ttl=60)  # 假设ttl是参数
    def expensive_func(x):
        nonlocal call_count
        call_count += 1
        return x * 2
    
    assert expensive_func(5) == 10
    assert call_count == 1  # 首次执行
    assert expensive_func(5) == 10
    assert call_count == 1  # 第二次未执行
    
    # 测试2:不同参数调用不同缓存
    assert expensive_func(3) == 6
    assert call_count == 2  # 新参数触发执行
    
    # 测试3:保留原始函数信息
    assert expensive_func.__name__ == "expensive_func"
    assert "x * 2" in expensive_func.__doc__

这个测试驱动开发(TDD)过程强制你思考: cache 必须接收 ttl 参数,必须基于参数生成缓存键,必须区分不同参数的调用。现在实现就简单了——照着测试写:

from functools import wraps
import time
from typing import Any, Callable, Dict, Optional

def cache(ttl: int = 300):
    def decorator(func: Callable) -> Callable:
        # 缓存存储:key=函数名+参数字符串,value=(结果, 时间戳)
        _cache: Dict[str, tuple] = {}
        
        @wraps(func)  # 关键!wraps必须在这里
        def wrapper(*args, **kwargs):
            # 生成缓存键:用repr确保参数可哈希,且能区分(1,)和1
            key = f"{func.__name__}:{repr(args)}:{repr(sorted(kwargs.items()))}"
            
            # 检查缓存是否有效
            if key in _cache:
                result, timestamp = _cache[key]
                if time.time() - timestamp < ttl:
                    return result
            
            # 执行原函数并缓存
            result = func(*args, **kwargs)
            _cache[key] = (result, time.time())
            return result
        
        # 添加清除缓存的方法,方便测试和调试
        wrapper.clear_cache = lambda: _cache.clear()
        return wrapper
    return decorator

注意 wrapper.clear_cache 这一行——这是实操中极有价值的技巧。很多装饰器缺少调试入口,导致缓存污染后只能重启服务。给 wrapper 动态添加方法,测试时调用 expensive_func.clear_cache() 即可重置,生产环境也能用 getattr(func, 'clear_cache', lambda: None)() 安全调用。

3.2 第二步:参数校验与默认值——拒绝“传啥都接”的脆弱设计

装饰器参数如果非法,应该在装饰时(即 @decorator(...) 执行阶段)就报错,而不是等到函数调用时才崩溃。比如 @retry(max_attempts=-1) 显然不合理,应该在 retry(-1) 阶段就抛出 ValueError

def retry(max_attempts: int = 3, delay: float = 1.0):
    if max_attempts < 1:
        raise ValueError(f"max_attempts must be >= 1, got {max_attempts}")
    if delay < 0:
        raise ValueError(f"delay must be >= 0, got {delay}")
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # ... 实现逻辑
        return wrapper
    return decorator

这种校验放在工厂函数里,比放在 wrapper 里更早暴露问题。用户写 @retry(max_attempts=0) 时,Python会立即报错,而不是等到某个API请求失败时才看到 ValueError 。同理,对于字符串参数,做 strip() 和非空检查;对于枚举类型,用 Enum Literal 限定取值范围。我在金融系统里见过一个 @validate(currency="USD") 装饰器,因为没校验 currency 是否在白名单里,导致黑客传入 "RUB" 绕过汇率校验——参数校验不是“防君子”,而是“防误操作”。

3.3 第三步:组合与顺序——理解装饰器栈的执行流

多个装饰器 @A @B @C 等价于 C = A(B(C)) ,即 从下往上装饰,从上往下执行 。这决定了组合逻辑:

@log
@retry(max_attempts=2)
@cache(ttl=60)
def api_call(url):
    return requests.get(url).json()

执行顺序是:

  1. api_call() 被调用 → 进入 log wrapper
  2. log.wrapper 调用 retry.wrapper
  3. retry.wrapper 调用 cache.wrapper
  4. cache.wrapper 检查缓存 → 命中则返回,未命中则调用 api_call 原函数

所以 cache 在最内层,负责最终执行; log 在最外层,负责最外层的日志。 关键推论 :如果 cache 要基于请求头(如 Authorization )生成缓存键,但 log url 改成了带时间戳的版本, cache 就永远无法命中。因此,组合时要考虑数据流:上游装饰器是否修改了下游依赖的参数?下游装饰器是否假设了上游的副作用?

解决方案是显式传递上下文。例如,让 log 把原始参数存到 kwargs 里:

def log(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 记录原始参数,供下游装饰器使用
        kwargs["_original_args"] = args
        kwargs["_original_kwargs"] = kwargs.copy()
        print(f"Calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} done")
        return result
    return wrapper

这样 cache 就可以读 kwargs.get("_original_args") 来生成键。但这增加了耦合,更好的方式是定义协议:所有装饰器都约定从 kwargs _context 字典,由最外层注入。

3.4 第四步:调试与可观测性——让装饰器“说话”

生产环境里,装饰器是黑盒,直到它出问题。我在一个电商项目里调试过一个 @rate_limit 装饰器,用户投诉“下单总是失败”,日志里却只有一句 Rate limit exceeded 。后来发现是 @auth 装饰器把 user_id 存在 request.user.id ,而 @rate_limit 却从 request.headers["X-User-ID"] 读——两者不一致。解决方法是在每个装饰器里加调试日志:

def rate_limit(calls: int = 10, period: int = 60):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 调试:打印关键决策依据
            user_id = get_user_id_from_context(*args, **kwargs)
            print(f"[DEBUG] rate_limit: user_id={user_id}, calls={calls}, period={period}")
            
            # 实际限流逻辑...
            if is_rate_limited(user_id, calls, period):
                print(f"[WARN] rate_limit blocked user {user_id}")
                raise RateLimitExceeded()
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

[DEBUG] 标签让运维能快速过滤日志; [WARN] 标签用于告警。更重要的是, 所有装饰器的调试日志格式统一 [模块名] 行为 描述 。这样用 grep "\[rate_limit\]" 就能隔离出所有限流日志,不用在千行日志里找关键词。我在团队推行这个规范后,平均故障定位时间从47分钟降到6分钟。

4. 高阶实战:解决五个真实世界中的装饰器难题

4.1 难题一:如何让装饰器兼容异步函数?

同步装饰器 @log func(*args, **kwargs) 调用,但异步函数要用 await func(*args, **kwargs) 。强行用同步装饰器套异步函数会报 RuntimeWarning: coroutine 'xxx' was never awaited 。解决方案是检测函数类型,动态选择调用方式:

import asyncio
from inspect import iscoroutinefunction

def log_async(func):
    @wraps(func)
    def sync_wrapper(*args, **kwargs):
        print(f"Sync calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"Sync {func.__name__} returned {result}")
        return result
    
    @wraps(func)
    async def async_wrapper(*args, **kwargs):
        print(f"Async calling {func.__name__}")
        result = await func(*args, **kwargs)
        print(f"Async {func.__name__} returned {result}")
        return result
    
    # 根据函数类型返回对应包装器
    if iscoroutinefunction(func):
        return async_wrapper
    else:
        return sync_wrapper

# 使用:自动适配
@log_async
def sync_func(x):
    return x + 1

@log_async
async def async_func(x):
    await asyncio.sleep(0.1)
    return x * 2

这个方案的关键是 iscoroutinefunction ,它比 inspect.isawaitable 更准确,因为后者判断的是返回值,而我们要判断的是函数本身是否是协程函数。注意: async_wrapper await func() 不能写成 await func(*args, **kwargs) ,因为 func 是协程函数,调用它返回协程对象, await 才执行。

4.2 难题二:如何避免装饰器污染全局状态?

常见错误是把缓存字典或计数器定义在模块顶层:

# ❌ 危险:所有被装饰函数共享同一份缓存
_cache = {}

def bad_cache(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        key = f"{func.__name__}:{args}"
        if key not in _cache:
            _cache[key] = func(*args, **kwargs)
        return _cache[key]
    return wrapper

问题在于: @bad_cache 装饰 func_a func_b 时,它们共用 _cache func_a 的缓存可能被 func_b 覆盖。正确做法是 每个装饰器实例独占状态

# ✅ 正确:闭包或类实例隔离状态
def good_cache(ttl=300):
    def decorator(func):
        # 每个func都有自己的_cache
        _cache = {}
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = f"{func.__name__}:{repr(args)}"
            # ... 缓存逻辑
        return wrapper
    return decorator

或者用类装饰器,状态在 self 上:

class GoodCache:
    def __init__(self, ttl=300):
        self.ttl = ttl
    
    def __call__(self, func):
        self._cache = {}  # 每个func实例一份
        @wraps(func)
        def wrapper(*args, **kwargs):
            # ... 
        return wrapper

4.3 难题三:如何为装饰器编写单元测试?

装饰器测试的核心是: 验证它是否正确修改了函数行为,且不破坏原始接口 。以 @retry 为例:

import pytest
from unittest.mock import Mock, patch

def test_retry_decorator():
    # 场景1:函数首次就成功
    mock_func = Mock(return_value="success")
    decorated = retry(max_attempts=3)(mock_func)
    
    result = decorated(1, 2)
    assert result == "success"
    assert mock_func.call_count == 1  # 只调用一次
    
    # 场景2:前两次失败,第三次成功
    mock_func.side_effect = [ValueError("fail1"), ValueError("fail2"), "success"]
    result = decorated(1, 2)
    assert result == "success"
    assert mock_func.call_count == 3  # 重试两次后成功
    
    # 场景3:全部失败,抛出最后一次异常
    mock_func.side_effect = [ValueError("fail1"), ValueError("fail2"), ValueError("fail3")]
    with pytest.raises(ValueError, match="fail3"):
        decorated(1, 2)
    assert mock_func.call_count == 3

def test_retry_preserves_signature():
    def original(a: int, b: str) -> bool:
        """Original docstring"""
        return True
    
    decorated = retry(max_attempts=2)(original)
    # 检查签名
    sig = inspect.signature(decorated)
    assert list(sig.parameters.keys()) == ["a", "b"]
    assert sig.return_annotation == bool
    # 检查文档
    assert decorated.__doc__ == "Original docstring"

测试覆盖了行为(重试逻辑)、接口(签名和文档)、边界(全部失败)。注意 Mock side_effect 可以是异常列表,完美模拟不稳定网络。

4.4 难题四:如何让装饰器支持类方法和静态方法?

@staticmethod @classmethod 会改变函数绑定方式,普通装饰器可能失效。例如:

class Calculator:
    @staticmethod
    @log
    def add(a, b):
        return a + b

@log 作用在 staticmethod 返回的对象上,但 staticmethod 对象不是普通函数, @wraps 可能失败。解决方案是让装饰器兼容描述符协议:

from functools import update_wrapper

def log(func):
    # 如果func是描述符(如staticmethod, classmethod),先unwrap
    if hasattr(func, '__func__'):
        # __func__ 属性存在于staticmethod和classmethod中
        raw_func = func.__func__
        wrapper = _make_wrapper(raw_func)
        # 重新包装成相同类型
        if isinstance(func, staticmethod):
            return staticmethod(wrapper)
        elif isinstance(func, classmethod):
            return classmethod(wrapper)
    else:
        wrapper = _make_wrapper(func)
    
    return wrapper

def _make_wrapper(func):
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} returned {result}")
        return result
    return update_wrapper(wrapper, func)

更简单的方法是: 要求用户把装饰器放在 @staticmethod 下方 (即 @log @staticmethod ),因为 @staticmethod 返回的是普通函数对象, @log 可以正常处理。这是社区通用做法,比自己实现描述符兼容更可靠。

4.5 难题五:如何在装饰器中安全访问请求上下文(如Flask/Gunicorn)?

Web框架中常需在装饰器里获取当前请求,但直接导入 flask.request 会报 RuntimeError: Working outside of application context 。正确方式是延迟获取,在 wrapper 中访问:

from flask import request, g

def require_auth(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # ✅ 在wrapper中访问,此时已有请求上下文
        auth_header = request.headers.get("Authorization")
        if not auth_header or not validate_token(auth_header):
            return {"error": "Unauthorized"}, 401
        
        # 将用户信息存入g,供视图函数使用
        g.current_user = get_user_from_token(auth_header)
        return func(*args, **kwargs)
    return wrapper

# 使用
@app.route("/api/data")
@require_auth
def get_data():
    return {"user": g.current_user.name}

关键点: request g 是线程/协程局部的,只有在请求处理过程中才有效。装饰器工厂函数( require_auth )在模块加载时执行,此时无上下文; wrapper 在每次请求时执行,上下文已就绪。切记不要在工厂函数里 request.headers ——那是灾难的开始。

5. 避坑指南:那些年我们踩过的装饰器深坑与实操心得

5.1 坑一: @wraps 位置错误——元数据丢失的隐形杀手

最常见的错误是把 @wraps 放在错误的位置。例如:

# ❌ 错误:wraps作用在decorator上,不是wrapper上
def log(func):
    @wraps(func)  # 这里错了!func是被装饰函数,但我们要包装wrapper
    def decorator(func):  # 这个func是参数,不是原始函数
        def wrapper(*args, **kwargs):
            ...
        return wrapper
    return decorator

正确写法必须是 @wraps(func) 作用于 wrapper 函数:

# ✅ 正确:wraps在wrapper定义处
def log(func):
    def decorator(func):
        @wraps(func)  # 作用于wrapper,复制func的元数据
        def wrapper(*args, **kwargs):
            ...
        return wrapper
    return decorator

实测心得:在PyCharm里按 Ctrl+Click 跳转到被装饰函数,如果跳转失败,90%是 wraps 位置错了。用 help(func) 检查 __doc__ 是否为空,是最快验证方式。

5.2 坑二:闭包变量修改—— nonlocal 的陷阱

想在装饰器里统计调用次数,有人这么写:

# ❌ 危险:count在闭包中,但多个装饰器实例共享
count = 0

def count_calls(func):
    def wrapper(*args, **kwargs):
        nonlocal count  # 这里nonlocal找不到count,因为count在模块层
        count += 1
        return func(*args, **kwargs)
    return wrapper

nonlocal 只能向上找嵌套作用域,模块层变量要用 global ,但 global 是全局共享的。正确做法是用可变对象(如字典)或类属性:

# ✅ 正确:每个装饰器实例有自己的计数器
def count_calls(func):
    counter = {"count": 0}  # 可变对象,闭包可修改
    @wraps(func)
    def wrapper(*args, **kwargs):
        counter["count"] += 1
        print(f"{func.__name__} called {counter['count']} times")
        return func(*args, **kwargs)
    wrapper.get_count = lambda: counter["count"]  # 暴露查询接口
    return wrapper

5.3 坑三:装饰器参数类型错误—— None 值引发的血案

当装饰器参数可选时,常犯错误是默认值为 None ,但在逻辑中直接使用:

# ❌ 危险:如果timeout=None,time.sleep(None)报错
def timeout(timeout_sec=None):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if timeout_sec is not None:  # 必须显式检查
                # 启动超时线程...
            return func(*args, **kwargs)
        return wrapper
    return decorator

更安全的方式是用哨兵对象(sentinel):

# ✅ 正确:用哨兵对象避免None歧义
_NO_TIMEOUT = object()

def timeout(timeout_sec=_NO_TIMEOUT):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if timeout_sec is not _NO_TIMEOUT:
                # 处理超时
                pass
            return func(*args, **kwargs)
        return wrapper
    return decorator

5.4 坑四:装饰器与类型提示冲突——mypy报错的根源

@wraps(func) 复制了 __annotations__ ,但如果 wrapper 的签名是 (*args, **kwargs) ,mypy 会认为返回值是 Any 。解决方案是用 typing.overload ParamSpec (Python 3.10+):

from typing import TypeVar, Callable, ParamSpec, Concatenate

P = ParamSpec('P')
R = TypeVar('R')

def log(func: Callable[P, R]) -> Callable[P, R]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        print(f"Calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} returned {result}")
        return result
    return wrapper

ParamSpec 保留了原始函数的参数类型,mypy 能正确推导 wrapper 的签名。对于老版本Python,用 @overload 声明多种签名,虽然繁琐但有效。

5.5 坑五:装饰器性能开销——不该有的循环引用

装饰器里如果意外创建循环引用,会导致内存泄漏。例如:

# ❌ 危险:wrapper闭包引用func,func又引用wrapper(如通过装饰器属性)
def log(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        ...
    wrapper.original_func = func  # func现在引用wrapper
    func.wrapper = wrapper         # wrapper引用func → 循环引用
    return wrapper

CPython 的垃圾回收器能处理,但PyPy或某些嵌入式环境可能不行。正确做法是用弱引用:

# ✅ 正确:用weakref避免循环引用
import weakref

def log(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        ...
    wrapper.original_func = weakref.ref(func)  # 弱引用
    return wrapper

最后分享一个个人体会:装饰器不是银弹。我在一个高并发日志系统里,曾用 @log 装饰所有函数,结果I/O阻塞导致TPS下降40%。后来改用结构化日志中间件,在WSGI层统一处理,性能提升3倍。所以, 装饰器的价值不在于“能做什么”,而在于“是否值得做” ——如果逻辑能抽成独立函数、中间件或配置项,就别硬塞进装饰器。它应该是锦上添花的利器,而不是掩盖架构缺陷的膏药。

更多推荐