Python装饰器

Python 装饰器:别被 @ 符号吓到,它就是个”包装盒”

第一次看到 @ 符号的时候,我也觉得像魔法——在函数上面加一行,行为就变了,完全不知道发生了什么。

后来明白了:装饰器就是个”包装盒”。你把一个函数放进去,它给你包装一下,加些功能,再还给你。函数还是那个函数,只是多了点本事。

装饰器到底是什么?三个例子看懂

例子1:函数可以赋值给变量

def greet(name):
    return f"Hello, {name}"

say_hello = greet   # 把函数赋值给变量
say_hello("Alice")  # "Hello, Alice"

函数跟数字、字符串一样,可以到处传递。

例子2:函数可以作为参数传给另一个函数

def apply(func, value):
    return func(value)

apply(greet, "Bob")   # "Hello, Bob"

例子3:函数可以返回另一个函数

def make_multiplier(factor):
    def multiplier(x):
        return x * factor
    return multiplier

double = make_multiplier(2)
double(5)   # 10

装饰器就是这三个能力的组合:接收一个函数,在它外面包一层新功能,返回包装后的函数。

def my_decorator(func):
    def wrapper():
        print("执行前")
        result = func()
        print("执行后")
        return result
    return wrapper

@my_decorator
def say_hello():
    print("Hello!")

say_hello()
# 执行前
# Hello!
# 执行后

@my_decorator 等价于 say_hello = my_decorator(say_hello)。就是把 say_hello 放进 my_decorator 里包装一下,再赋值回来。

基础装饰器:给函数加”前后置”逻辑

最常用的场景:在函数执行前后自动做点事——记日志、计时、权限检查。

import time
import functools

def timer(func):
    @functools.wraps(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

@timer
def slow_task():
    time.sleep(1)
    return "完成"

slow_task()   # slow_task 耗时: 1.0002s

*args, **kwargs 是为了让装饰器能接受任意参数的函数——不管你传什么,我都原样传进去。

为什么一定要加 @functools.wraps

不加的话,被装饰函数的”身份信息”会丢失:

def bad_decorator(func):
    def wrapper():
        return func()
    return wrapper

@bad_decorator
def greet():
    """打招呼"""
    pass

greet.__name__   # 'wrapper',不是 'greet'
greet.__doc__    # None,文档丢了

调试的时候,堆栈里全是 wrapper,根本不知道是哪个函数出问题。用 @functools.wraps 解决:

import functools

def good_decorator(func):
    @functools.wraps(func)   # 这行把 greet 的名字、文档、签名都复制到 wrapper 上
    def wrapper():
        return func()
    return wrapper

@good_decorator
def greet():
    """打招呼"""
    pass

greet.__name__   # 'greet' 
greet.__doc__    # '打招呼' 

写装饰器的铁律:每个装饰器里都用 @functools.wraps

带参数的装饰器:三层嵌套

上面的装饰器都是”固定行为”——计时就是计时,不让你配置。

如果需要配置参数,比如”重试3次”、”重试间隔2秒”,就需要三层嵌套:

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts:
                        raise
                    print(f"第{attempt}次失败,{delay}秒后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, delay=1)
def unstable_request():
    # 可能失败的操作
    pass

三层各管一摊:

层级 作用 参数
最外层 retry 接收配置参数 max_attemptsdelay
中间层 decorator 接收被装饰的函数 func
最内层 wrapper 执行实际逻辑 *args, **kwargs

这个模式看着复杂,但写多了就熟了——外层接配置,中层接函数,内层接调用参数。

实际场景:这几种装饰器最常用

场景1:计时器(监控接口性能)

@timer
def query_database():
    # 慢查询
    pass

API 响应慢的时候,加上 @timer 就能知道每个函数耗时多少,快速定位瓶颈。

场景2:重试(调用外部服务)

@retry(max_attempts=3, delay=1)
def call_third_party_api():
    # 网络请求可能失败
    pass

外部服务不稳定的时候,自动重试比人工介入快多了。

场景3:缓存(避免重复计算)

@lru_cache(maxsize=128)
def expensive_compute(n):
    # 耗时计算
    pass

functools.lru_cache 是 Python 内置的缓存装饰器,同一个参数只算一次。

场景4:权限检查(Web 开发)

@login_required
def dashboard():
    # 只有登录用户能访问
    pass

Flask、FastAPI 里这类装饰器随处可见。

场景5:日志(调试生产问题)

@log_call
def process_order(order_id):
    # 自动记录入参和返回值
    pass

出问题的时候看日志就知道”哪个函数被调了、传了什么参数、返回了什么”。

多个装饰器叠加:顺序怎么算

@decorator_a
@decorator_b
@decorator_c
def func():
    pass

# 等价于:
func = decorator_a(decorator_b(decorator_c(func)))

执行顺序:从下往上包装,从上往下执行。

@log
@timer
def work():
    pass

# work() 执行时:
# log 的前置 → timer 的前置 → work → timer 的后置 → log 的后置

想清楚包装顺序再叠加,不然行为可能跟你预期的不一样。

类装饰器:需要状态的时候用

函数装饰器每次调用都是独立的,如果需要”累计调用次数”这种状态,用类装饰器。

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0
    
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"调用了 {self.count} 次")
        return self.func(*args, **kwargs)

@CountCalls
def greet():
    print("Hello")

greet()   # 调用了 1 次
greet()   # 调用了 2 次
greet()   # 调用了 3 次

__call__ 让类的实例可以像函数一样被调用。CountCalls 的实例 greet 被调用时,走的是 __call__ 方法。

异步装饰器:用于 async def

如果你的函数是 async def,装饰器也必须是异步的:

def async_timer(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = await func(*args, **kwargs)   # 注意这里用 await
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} 耗时: {elapsed:.4f}s")
        return result
    return wrapper

@async_timer
async def fetch():
    await asyncio.sleep(1)
    return "数据"

如果装饰器想同时支持同步和异步,用 inspect.iscoroutinefunction(func) 判断,写两个分支。

一个完整的实战:带超时的重试装饰器

import asyncio
import functools
import time

def retry_with_timeout(max_attempts=3, timeout=5):
    """
    重试装饰器,支持超时控制
    """
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return await asyncio.wait_for(
                        func(*args, **kwargs),
                        timeout=timeout
                    )
                except asyncio.TimeoutError:
                    print(f"第{attempt}次超时")
                    if attempt == max_attempts:
                        raise
                except Exception as e:
                    print(f"第{attempt}次失败: {e}")
                    if attempt == max_attempts:
                        raise
                    await asyncio.sleep(1)
        return wrapper
    return decorator

@retry_with_timeout(max_attempts=3, timeout=2)
async def fetch_data():
    # 可能超时或失败的操作
    await asyncio.sleep(3)  # 模拟慢请求
    return "数据"

这个装饰器解决了一个真实问题:外部 API 可能超时,也可能临时报错,自动重试+超时控制比手工处理可靠得多。