Python装饰器:怎么让代码在不改变结构的情况下获得新能力

你在维护一个线上服务，某天产品经理过来说：「给那个查询接口加一个权限校验，用户没登录的一律拒绝。」你找到那个函数，发现它已经被其他地方引用了三十多处。更麻烦的是，它不是你写的，你甚至不确定改动会不会影响已有的调用逻辑。

这种场景在工程中极其常见。修改原函数的风险太高，但不加功能又不行。Python 给出的答案是：装饰器。

装饰器的核心场景：不改源码，叠加能力

想象一个函数：

def get_user_data(user_id):    return db.query(user_id)

产品经理要求加权限校验，最朴素的做法是改函数体：

def get_user_data(user_id):    if not current_user.is_authenticated:        raise PermissionError    return db.query(user_id)

问题是：这个函数在三十个地方被调用，你改一次，所有调用点都算改过。代码评审时 reviewer 很难判断「这次改动是加校验还是顺便改了别的逻辑」。

装饰器提供了另一种思路：在函数的外层包一层逻辑，原函数保持不动。

def require_auth(func):    def wrapper(user_id):        if not current_user.is_authenticated:            raise PermissionError        return func(user_id)    return wrapper@require_authdef get_user_data(user_id):    return db.query(user_id)

被 @require_auth 装饰的 get_user_data，对外的调用方式完全没有变化，但行为已经增加了权限校验。用 Git diff 来看，只多了一行 @require_auth，review 成本极低。

同样的思路可以扩展到日志记录：

def log_calls(func):    def wrapper(*args, **kwargs):        start = time.time()        result = func(*args, **kwargs)        print(f'{func.__name__} took {time.time() - start:.3f}s')        return result    return wrapper

以及性能计时：

import timedef timing(func):    def wrapper(*args, **kwargs):        start = time.perf_counter()        result = func(*args, **kwargs)        elapsed = time.perf_counter() - start        print(f'{func.__name__}: {elapsed:.4f}s')        return result    return wrapper

三个能力（权限校验、日志、计时），对应三个装饰器，原函数代码零改动。这才是装饰器的工程价值：横向扩展，而非纵向侵入。

装饰器的本质：函数是对象，返回函数的高阶函数

理解装饰器不需要任何魔法语法，核心只需要一个认知：在 Python 中，函数是对象。

def greet(name):    return f'Hello, {name}'print(greet)          # <function greet at 0x7f9b2c3d44c0>print(greet.__name__) # greet

既然是对象，就可以作为参数传入另一个函数，也可以被另一个函数作为返回值。装饰器本质上就是一个接收函数、返回函数的高阶函数。

def my_decorator(func):   # func 是被装饰的函数    def inner(*args, **kwargs):        result = func(*args, **kwargs)        return result    return inner          # 返回一个新函数

@my_decorator 写在某个函数定义前，就等于：

def some_function():    passsome_function = my_decorator(some_function)

这就是装饰器的全部语法糖。理解了这个等价关系，就不会再被 @ 符号迷惑。

@functools.wraps：为什么必须用它

不加上 @functools.wraps 的装饰器，会产生一种微妙但危险的 bug——函数元信息丢失。

def bad_decorator(func):    def wrapper(*args, **kwargs):        return func(*args, **kwargs)    return wrapper@bad_decoratordef fetch_data():    '''获取远程数据'''    passprint(fetch_data.__name__)   # wrapperprint(fetch_data.__doc__)    # None

fetch_data.__name__ 变成了 wrapper，fetch_data.__doc__ 变成了 None。这会导致什么问题？

pydoc、IDE 自动补全、内省（introspection）全部失效。 如果你的装饰器用在框架中间件上，框架依赖函数签名做路由判断，也会出错。更隐蔽的是日志——你看到一行日志写着「wrapper 执行出错」，根本不知道是哪个原始函数出的问题。

解决方法是加一行：

import functoolsdef good_decorator(func):    @functools.wraps(func)   # 关键这一行    def wrapper(*args, **kwargs):        return func(*args, **kwargs)    return wrapper@good_decoratordef fetch_data():    '''获取远程数据'''    passprint(fetch_data.__name__)   # fetch_dataprint(fetch_data.__doc__)    # 获取远程数据

@functools.wraps(func) 本质上是在 wrapper 上复制了原函数 func 的 __name__、__doc__、__module__、__annotations__ 等属性。这不是可选项，而是装饰器的标准规范——任何不接受 functools.wraps 的装饰器实现都是不完整的。

带参数的装饰器：三层闭包结构

基础的装饰器没有参数。给装饰器本身加参数，需要再多一层函数，形成三层闭包：

def retry(max_attempts=3, delay=1.0):    def decorator(func):        @functools.wraps(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                    time.sleep(delay)        return wrapper    return decorator

这里 retry 是最外层函数，接收装饰器参数；decorator 是中间层，接收被装饰函数；wrapper 是最内层，执行实际逻辑。

使用方式：

@retry(max_attempts=5, delay=0.5)def call_api(url):    response = requests.get(url)    return response.json()

这个结构容易犯的一个错误是在内层函数中错误地引用外层变量。Python 的闭包遵循 LEGB 规则（Local → Enclosing → Global → Built-in），但内层函数对外层变量的引用是只读的，如果你试图在 wrapper 里修改 max_attempts，不会修改外层的 max_attempts，而是创建一个新的局部变量。这在简单场景下不会出问题，但在需要计数状态的装饰器中，常见解法是用 nonlocal 声明或者把状态放在一个可变容器里。

一个带状态的缓存装饰器：

def lru_cache(maxsize=128):    def decorator(func):        cache = {}        @functools.wraps(func)        def wrapper(*args, **kwargs):            key = (args, tuple(sorted(kwargs.items())))            if key in cache:                return cache[key]            result = func(*args, **kwargs)            if len(cache) >= maxsize:                cache.pop(next(iter(cache)))            cache[key] = result            return result        return wrapper    return decorator

这里用字典 cache 存储调用结果，key 的构建方式需要特别说明——*args 是可哈希的，但 **kwargs 的字典顺序不固定，所以需要用 tuple(sorted(kwargs.items())) 规范化。缓存淘汰策略也故意简化了，实际工程中可以考虑 collections.OrderedDict 或 functools.lru_cache（标准库已提供）。

类装饰器：__call__ 的使用时机

装饰器不一定是函数，也可以是类。当装饰器需要保存状态时，类比函数更自然：

class RateLimiter:    def __init__(self, max_calls, period):        self.max_calls = max_calls        self.period = period        self.calls = []    def __call__(self, func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            now = time.time()            self.calls = [t for t in self.calls if now - t < self.period]            if len(self.calls) >= self.max_calls:                raise RuntimeError('Rate limit exceeded')            self.calls.append(now)            return func(*args, **kwargs)        return wrapper

__call__ 使得类的实例可以像函数一样被调用。self.__init__ 中的参数就是装饰器参数（max_calls、period），实例本身作为装饰器使用，__call__ 返回的 wrapper 才是真正的调用目标。

使用场景上，类装饰器比函数装饰器更适合：带状态（限流、计数）、需要初始化复杂资源（数据库连接、API 客户端）的场景。函数装饰器则更适合无状态、一行搞定的情况。

两种实现可以互相替代，但语义不同。选择类还是函数，取决于你需要的复杂度。

装饰器的顺序问题：谁在内层，谁先执行

多个装饰器可以叠加在同一函数上：

@log_calls@timing@require_authdef get_data():    pass

执行顺序是从下往上：先 require_auth，再 timing，再 log_calls。

理解方式很简单——把装饰器展开成赋值语句：

def get_data():    passget_data = require_auth(get_data)   # 第一步：先应用 require_authget_data = timing(get_data)          # 第二步：再应用 timingget_data = log_calls(get_data)      # 第三步：最后应用 log_calls

所以最外层（@log_calls）对应最先执行的包装逻辑。

实际调试中，如果发现执行顺序混乱，可以加一行日志在每个 wrapper 里打印函数名：

def debug_decorator(func):    @functools.wraps(func)    def wrapper(*args, **kwargs):        print(f'Entering {func.__name__}')        result = func(*args, **kwargs)        print(f'Exiting {func.__name__}')        return result    return wrapper

装饰器的顺序错误通常是排查生产 Bug 时最难定位的问题之一，因为它涉及的是调用链的编排，而不是单个函数的逻辑。养成写装饰器时打印函数名的习惯，可以在出问题时有迹可循。

实战：完整重试装饰器（指数退避） + 缓存装饰器

指数退避重试装饰器

指数退避（exponential backoff）的逻辑是：每次重试前等待的时间按指数增长，同时加上随机抖动避免惊群效应。

import functoolsimport timeimport randomdef retry_with_backoff(max_attempts=3, base_delay=1.0, max_delay=60.0, jitter=True):    '''    指数退避重试装饰器    参数:        max_attempts: 最大尝试次数（含首次）        base_delay: 基础等待时间（秒）        max_delay: 最大等待时间（秒）        jitter: 是否添加随机抖动    '''    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            last_exception = None            for attempt in range(max_attempts):                try:                    return func(*args, **kwargs)                except Exception as e:                    last_exception = e                    if attempt == max_attempts - 1:                        raise last_exception                    wait_time = min(base_delay * (2 ** attempt), max_delay)                    if jitter:                        wait_time *= (0.5 + random.random())  # 0.5x ~ 1.5x                    print(f'Attempt {attempt + 1} failed: {e}. '                          f'Retrying in {wait_time:.2f}s...')                    time.sleep(wait_time)            raise last_exception        return wrapper    return decorator

使用示例：

@retry_with_backoff(max_attempts=4, base_delay=0.5, max_delay=10.0)def unreliable_network_call():    if random.random() < 0.7:        raise ConnectionError('Network error')    return 'success'

注意实现中两个关键细节：保留原始异常（raise last_exception 而非 raise），这样堆栈跟踪不会被吞掉；抖动系数用 0.5 + random.random() 产生 0.5x 到 1.5x 的随机倍数，而不是直接用 random.random() 乘以等待时间——这样保证即使 jitter 了，基本的指数增长趋势依然保持。

缓存装饰器

import functoolsimport hashlibimport pickledef cached(func):    '''    基于函数参数哈希的缓存装饰器    '''    @functools.wraps(func)    def wrapper(*args, **kwargs):        try:            key_args = pickle.dumps((args, sorted(kwargs.items())))            cache_key = hashlib.md5(key_args).hexdigest()        except Exception:            return func(*args, **kwargs)        if not hasattr(wrapper, '_cache'):            wrapper._cache = {}        if cache_key in wrapper._cache:            return wrapper._cache[cache_key]        result = func(*args, **kwargs)        wrapper._cache[cache_key] = result        return result    return wrapper

注意这里用 pickle 序列化参数来生成缓存 key。pickle.dumps 可能失败（比如参数包含无法序列化的对象），所以加了 try-except，失败时走正常调用路径而不是直接报错。

对于需要严格缓存管理的场景，可以给装饰器加上 TTL（过期时间）或最大容量限制：

def cached_ttl(ttl_seconds=300, max_entries=1000):    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            try:                key_data = pickle.dumps((args, sorted(kwargs.items())))                cache_key = hashlib.md5(key_data).hexdigest()            except Exception:                return func(*args, **kwargs)            if not hasattr(wrapper, '_cache'):                wrapper._cache = {}                wrapper._cache_time = {}            now = time.time()            if cache_key in wrapper._cache:                if now - wrapper._cache_time[cache_key] < ttl_seconds:                    return wrapper._cache[cache_key]            result = func(*args, **kwargs)            wrapper._cache[cache_key] = result            wrapper._cache_time[cache_key] = now            if len(wrapper._cache) > max_entries:                oldest_key = min(wrapper._cache_time, key=wrapper._cache_time.get)                del wrapper._cache[oldest_key]                del wrapper._cache_time[oldest_key]            return result        return wrapper    return decorator

标准库的 functools.lru_cache 在大多数场景下是更好的选择（用 C 实现，性能更高，支持 cache_clear）。这里的自定义实现主要用于理解原理，或者需要特定淘汰策略时作为参考。

装饰器的本质是一个强有力的约定：函数的输入输出不变，内部逻辑可以任意扩展。这个约定让 AOP（面向切面编程）在 Python 中以极低的成本实现，也让它成为框架和中间件设计的基石。理解它不需要深入元编程，只需要理解「函数是对象，返回函数的高阶函数」这一个问题。