Python装饰器本质与实战:从函数工厂到生产级应用
Python 装饰器(Decorators)这个概念,我第一次真正搞懂它,不是在看官方文档的时候,而是在调试一个连续三天都报TypeError: 'NoneType' object is not callable的函数时——那个函数被我“顺手”加了个没返回值的装饰器,结果整个调用链无声崩塌。后来我翻了十几份教程,发现绝大多数人讲装饰器,要么卡在“语法糖”三个字上打转,要么堆砌五层嵌套函数让人头晕目眩。其实它根本没那么玄:装饰器就是个“函数加工厂”——你把一个函数送进去,它给你吐出一个功能增强版的新函数,原函数纹丝不动,连内存地址都不变,但行为已经悄悄升级了。这个机制在日志记录、权限校验、缓存控制、性能监控、参数预处理等场景里天天被用,不是炫技,是真正在解决重复劳动问题。如果你写过 Flask 的@app.route()、Django 的@login_required,或者用过@lru_cache加速递归,那你已经和装饰器打过照面了——只是可能还不知道它长什么样、怎么自己造一个、为什么非得这么写。这篇内容不讲抽象定义,不列教科书式流程图,只讲我在真实项目里怎么设计、怎么调试、怎么绕开那些坑,以及为什么某些写法看似简洁实则埋雷。适合刚学完函数和闭包、正卡在“@xxx到底替换了什么”的人;也适合写了两年 Python 却从没自己写过装饰器的开发者——因为很多团队里,装饰器是资深工程师才碰的“高危区”,但其实只要理解三件事:函数是一等公民、闭包能保存状态、@符号只是语法糖,你就已经站在了门口。
1. 装饰器的本质解构与设计逻辑
1.1 它不是魔法,而是“函数的函数”的自然延伸
很多人一看到装饰器就下意识觉得“高级”“难懂”,其实是被@符号带偏了。我们先彻底扔掉这个符号,回到最原始的调用方式。假设你有一个普通函数:
def greet(name): return f"Hello, {name}!"现在你想给它加个功能:每次调用前打印一句“Calling greet…”。最直白的做法是改函数体:
def greet(name): print("Calling greet...") return f"Hello, {name}!"但这违反了“开闭原则”——你要为每个需要日志的函数都手动加一行print,而且一旦要删日志,又得逐个改。更好的办法是写一个“日志生成器”,它接收任意函数,返回一个带日志的新函数:
def add_logging(func): def wrapper(*args, **kwargs): print(f"Calling {func.__name__} with args={args}, kwargs={kwargs}") result = func(*args, **kwargs) print(f"{func.__name__} returned: {result}") return result return wrapper注意这里的关键点:add_logging本身不执行任何业务逻辑,它只做一件事——接收一个函数func,然后定义并返回另一个函数wrapper。这个wrapper函数内部调用了func,并在调用前后插入了日志逻辑。此时,你可以这样使用它:
greet_logged = add_logging(greet) greet_logged("Alice") # 输出日志 + Hello, Alice!这完全不依赖@,就是一个标准的高阶函数调用。add_logging是“函数的函数”,wrapper是闭包,它记住了外层传入的func。这就是装饰器的全部骨架:输入函数 → 返回新函数 → 新函数增强原逻辑。@只是让这个过程更简洁:
@add_logging def greet(name): return f"Hello, {name}!"等价于:
def greet(name): return f"Hello, {name}!" greet = add_logging(greet)编译器在解析@add_logging时,做的就是最后这行赋值。所以@不是语法魔法,是语法糖,是 Python 为你自动写的那行greet = add_logging(greet)。理解这一点,就拆掉了第一层心理障碍。
1.2 为什么必须用闭包?不能直接在add_logging里执行?
有人会问:既然add_logging接收了func,为什么不在它里面直接调用func,而要再套一层wrapper?比如这样写:
def bad_add_logging(func): print(f"About to call {func.__name__}") result = func() # ❌ 错误!func 参数未知,且只执行一次 print(f"Got result: {result}") return result # ❌ 返回的是结果,不是函数!这完全错了。原因有三:
第一,装饰器的职责是“准备”,不是“执行”。你装饰一个函数,不是为了让它立刻运行,而是为了在将来每次调用它时都带上增强逻辑。bad_add_logging在定义阶段就强行执行了func(),这既不符合预期(你还没传参呢),又破坏了函数的可复用性。
第二,函数签名丢失。greet(name)需要一个name参数,但bad_add_logging里硬编码了func(),没有*args, **kwargs,调用时必然报错。
第三,返回值类型错误。装饰器必须返回一个可调用对象(callable),通常是函数。bad_add_logging返回的是result(比如字符串"Hello, Alice!"),它不可调用,后续greet_logged("Bob")就会报TypeError: 'str' object is not callable。
闭包wrapper解决了所有问题:它不立即执行,它保存了func的引用,它用*args, **kwargs兼容任意签名,它自己是一个函数,可以被反复调用。这是 Python 函数式编程的基石能力,不是装饰器独有,而是所有高阶函数设计的通用范式。
1.3 带参数的装饰器:为什么需要“三层函数”?
现实项目中,你经常需要配置装饰器的行为。比如日志装饰器,有时只想记录函数名,有时还要记录耗时,有时要开关日志级别。这就引出了带参数的装饰器,比如:
@add_logging(level="DEBUG") def greet(name): return f"Hello, {name}!"这看起来像是装饰器接收了参数,但注意:@add_logging(level="DEBUG")括号里有参数,说明它不是一个函数,而是一个调用表达式。Python 规定,带参数的装饰器必须是“返回装饰器的函数”。也就是说,add_logging(level="DEBUG")这个表达式的结果,必须是一个符合前面定义的、接收函数并返回新函数的装饰器。
因此,它的结构必然是三层:
def add_logging(level="INFO"): # 第一层:接收装饰器参数(如 level) def decorator(func): # 第二层:接收被装饰的函数 def wrapper(*args, **kwargs): # 第三层:实际执行逻辑,可访问 level 和 func if level == "DEBUG": print(f"[DEBUG] Calling {func.__name__}...") result = func(*args, **kwargs) return result return wrapper return decorator # 返回第二层函数为什么不能两层?试试看:
# ❌ 错误尝试:试图在 decorator 内部直接用 level def bad_add_logging(level="INFO", func=None): if func is None: return functools.partial(bad_add_logging, level=level) def wrapper(*args, **kwargs): print(f"[{level}] Calling {func.__name__}...") return func(*args, **kwargs) return wrapper这种写法看似聪明,但破坏了装饰器的语义一致性。@add_logging("DEBUG")要求add_logging("DEBUG")必须返回一个装饰器(即接收func的函数),而functools.partial返回的是一个部分应用对象,不是函数类型,某些框架或静态检查工具会报错。更重要的是,它让代码变得难以推理——你得记住partial的行为,而不是清晰地看到“参数层 → 装饰器层 → 执行层”的数据流。三层结构虽然多写几行,但每一层职责单一、意图明确,是经过大量工程实践验证的稳健模式。
2. 核心细节解析与实操要点
2.1*args, **kwargs不是摆设,是签名兼容的生命线
几乎所有教程都会告诉你装饰器里要用*args, **kwargs,但很少解释为什么它如此关键,以及漏掉它会引发什么具体问题。我们来看一个真实踩坑案例。
假设你写了一个简单的计时装饰器:
def timer(func): def wrapper(): start = time.time() result = func() end = time.time() print(f"{func.__name__} took {end - start:.4f}s") return result return wrapper然后你用它装饰一个带参数的函数:
@timer def calculate(a, b): time.sleep(0.1) return a + b calculate(3, 5) # TypeError: wrapper() takes 0 positional arguments but 2 were given错误信息非常清楚:wrapper()不接受参数,但你传了两个。这是因为wrapper的定义是def wrapper():,它硬编码了零参数。而calculate(3, 5)调用时,Python 试图把3, 5传给wrapper,但wrapper拒绝接收。
解决方案就是*args, **kwargs:
def timer(func): def wrapper(*args, **kwargs): # ✅ 接收任意位置和关键字参数 start = time.time() result = func(*args, **kwargs) # ✅ 原样传给原函数 end = time.time() print(f"{func.__name__} took {end - start:.4f}s") return result return wrapper这里有两个*args, **kwargs,它们的作用完全不同:
def wrapper(*args, **kwargs):声明wrapper自己能接收任意参数。func(*args, **kwargs):将接收到的参数,原封不动地解包传递给func。
这保证了装饰后的函数和原函数拥有完全一致的调用接口。用户不需要知道这个函数被装饰过,调用方式丝毫不变。这是装饰器能被大规模采用的前提——它对使用者透明。
提示:如果你的装饰器需要修改参数(比如统一添加一个
user_id),可以在wrapper内部操作args或kwargs,然后再传给func。例如:kwargs["user_id"] = get_current_user_id()。但务必确保不破坏原函数的语义。
2.2functools.wraps:不是可选项,是职业素养的底线
写完一个装饰器,你可能会发现一些奇怪现象:
@timer def greet(name): """Say hello to someone""" return f"Hello, {name}!" print(greet.__name__) # 输出:wrapper print(greet.__doc__) # 输出:Nonegreet的名字变成了wrapper,文档字符串丢失了。这是因为greet现在指向的是wrapper函数,而wrapper是你定义的内部函数,它当然没有greet的元信息。
这在日常开发中可能只是个小 annoyance,但在以下场景会引发严重问题:
- 调试困难:日志里打印
wrapper而不是greet,你得层层扒源码才能定位。 - IDE 支持失效:PyCharm、VS Code 无法正确跳转到原函数定义,参数提示、类型推导全乱。
- 框架集成失败:Flask、FastAPI 等框架依赖函数的
__name__和__doc__生成 API 文档或路由信息。如果装饰后__name__变成wrapper,Swagger 页面可能显示错误的端点名。 - 单元测试崩溃:有些测试框架通过函数名匹配测试用例,名字变了就找不到对应测试。
functools.wraps就是为解决这个问题而生的。它是一个装饰器工厂,作用是将原函数的元信息(__name__,__doc__,__module__,__annotations__,__dict__等)复制到包装函数上:
from functools import wraps def timer(func): @wraps(func) # ✅ 关键一行 def wrapper(*args, **kwargs): start = time.time() result = func(*args, **kwargs) end = time.time() print(f"{func.__name__} took {end - start:.4f}s") return result return wrapper现在再测试:
print(greet.__name__) # 输出:greet ✅ print(greet.__doc__) # 输出:Say hello to someone ✅@wraps(func)的本质是调用update_wrapper(wrapper, func),它内部做了大量属性拷贝工作。这不是“锦上添花”,而是编写生产级装饰器的强制要求。我见过太多团队因为省了这一行,导致线上 API 文档生成失败,排查了两天才发现是装饰器没加@wraps。把它当成和import语句一样,写装饰器时第一反应就加上,形成肌肉记忆。
2.3 类装饰器:当函数不够用时的务实选择
到目前为止,所有例子都是函数装饰器。但 Python 还支持类装饰器,即用一个类的实例来充当装饰器。它的语法是:
@MyClassDecorator() def my_function(): pass类装饰器的核心在于:该类必须实现__call__方法,因为装饰器最终要被调用(decorator(func)),而只有实现了__call__的对象才是可调用的。
一个典型的类装饰器用于统计函数调用次数:
class CallCounter: def __init__(self): self.count = 0 def __call__(self, func): @wraps(func) def wrapper(*args, **kwargs): self.count += 1 print(f"{func.__name__} has been called {self.count} times") return func(*args, **kwargs) return wrapper # 使用 counter = CallCounter() @counter def greet(name): return f"Hello, {name}!"这里counter是一个CallCounter实例,@counter等价于greet = counter(greet)。由于CallCounter有__call__方法,所以counter(greet)是合法的,它返回wrapper。
类装饰器的优势在于:它可以维护状态(state)。上面的self.count就是跨多次调用的共享状态。而函数装饰器如果想维护状态,通常得用闭包变量或全局变量,不如类清晰。
但类装饰器也有明显缺点:
- 复杂度更高:需要理解
__init__,__call__, 以及实例生命周期。 - 状态隔离问题:
counter是单例,所有被它装饰的函数共享同一个count。如果你想为每个函数单独计数,就得在__call__内部用字典按funcID 存储,代码立刻变臃肿。 - 继承与复用困难:相比函数,类的组合、继承在装饰器场景下并不常见,反而增加理解成本。
我的经验是:90% 的场景,函数装饰器足够且更优;只有当你明确需要跨调用的状态管理,且该状态逻辑较重时,才考虑类装饰器。比如一个连接池装饰器,需要管理多个数据库连接的创建、复用、回收,用类封装状态和方法会更自然。但对于日志、计时、缓存这类无状态或轻状态的增强,坚持用函数装饰器,代码更短、更易读、更易测。
3. 实操过程与核心环节实现
3.1 从零开始:手写一个带配置的认证装饰器
我们来实战一个真实项目中高频使用的装饰器:API 认证检查。假设你的 Web API 要求所有请求携带有效的AuthorizationHeader,格式为Bearer <token>,且 token 必须在 Redis 中存在且未过期。我们将一步步构建一个灵活、可配置、可测试的装饰器。
第一步:定义基础骨架
先不考虑 Redis,只做最简验证:
from functools import wraps from typing import Callable, Any def require_auth(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs): # TODO: 从请求中提取 token 并验证 return func(*args, **kwargs) return wrapper这是一个空壳,但已具备装饰器的所有基本要素:接收func,返回wrapper,用@wraps保持元信息。
第二步:注入依赖与配置
硬编码 Redis 连接是反模式。我们需要让装饰器能接收配置,比如 Redis 客户端实例、token 前缀、过期时间等。这就要用到带参数的三层结构:
def require_auth( redis_client=None, token_prefix: str = "Bearer ", timeout_seconds: int = 3600 ) -> Callable: # 第一层:接收装饰器配置参数 def decorator(func: Callable) -> Callable: # 第二层:接收被装饰函数 @wraps(func) def wrapper(*args, **kwargs): # 第三层:执行逻辑,可访问所有配置参数 # 但这里还缺关键一步:如何获取当前请求? # 在 Web 框架中,请求对象通常作为第一个参数传入 # 我们约定:被装饰的函数第一个参数是 request 对象 if not args: raise ValueError("require_auth expects first argument to be request") request = args[0] auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith(token_prefix): raise PermissionError("Invalid or missing Authorization header") token = auth_header[len(token_prefix):].strip() # TODO: 用 redis_client 验证 token return func(*args, **kwargs) return wrapper return decorator注意这里我们做了几个关键设计决策:
- 显式依赖注入:
redis_client作为参数传入,而不是在函数内import redis; redis.Redis()。这使得装饰器可测试(你可以传入 Mock 客户端),也符合依赖倒置原则。 - 合理默认值:
token_prefix默认"Bearer ",timeout_seconds默认3600(1小时),用户不传时也能工作。 - 清晰的错误处理:当缺少请求参数或 Header 格式错误时,抛出明确异常,而不是静默失败。
第三步:集成 Redis 验证与异常处理
现在补全 Redis 部分。我们假设redis_client有get方法,返回bytes或None:
def require_auth( redis_client=None, token_prefix: str = "Bearer ", timeout_seconds: int = 3600 ) -> Callable: def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs): if not args: raise ValueError("require_auth expects first argument to be request") request = args[0] auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith(token_prefix): raise PermissionError("Invalid or missing Authorization header") token = auth_header[len(token_prefix):].strip() if not token: raise PermissionError("Empty token") # 验证 token 是否存在于 Redis try: user_id_bytes = redis_client.get(f"auth:{token}") if user_id_bytes is None: raise PermissionError("Invalid or expired token") # 可选:刷新过期时间 redis_client.expire(f"auth:{token}", timeout_seconds) # 将 user_id 注入到 kwargs,供业务函数使用 user_id = user_id_bytes.decode("utf-8") kwargs["user_id"] = user_id except Exception as e: # 不暴露底层错误细节给客户端 raise PermissionError("Authentication failed") from e return func(*args, **kwargs) return wrapper return decorator这里我们增加了:
- 空 token 检查:防止
""被当作有效 token。 - Redis 异常捕获:
redis_client.get可能因网络问题抛出异常,我们统一转为PermissionError,避免泄露 Redis 细节。 - 用户 ID 注入:
kwargs["user_id"] = user_id,这样业务函数可以直接用user_id,无需自己解析 token。这是装饰器提供价值的关键——它不只是检查,还预处理了数据。
第四步:使用示例与完整调用链
假设你用的是 FastAPI:
from fastapi import FastAPI, Request, HTTPException import redis app = FastAPI() redis_client = redis.Redis(host="localhost", port=6379, db=0) @app.get("/profile") @require_auth(redis_client=redis_client) def get_profile(request: Request, user_id: str): # user_id 已由装饰器注入,无需自己解析 return {"user_id": user_id, "data": "profile_data"}调用流程是:
- 客户端发请求:
GET /profile,Header:Authorization: Bearer abc123 - FastAPI 将
Request对象作为第一个参数传给get_profile require_auth的wrapper拦截调用,提取abc123,查 Redis 得到user_id="u1001"wrapper将user_id="u1001"加入kwargswrapper调用get_profile(request, user_id="u1001")get_profile业务逻辑执行,返回结果
整个过程对get_profile函数完全透明,它只关心自己的业务,认证逻辑被完美剥离。
3.2 装饰器的单元测试:为什么 mock 比你想象的更重要
写完装饰器,不测试等于没写。装饰器的测试难点在于:它改变了函数的行为,但你不能(也不应该)在测试中启动真实的 Redis 或 Web 服务器。我们必须用unittest.mock来隔离外部依赖。
我们为require_auth写一个完整的测试用例:
import pytest from unittest.mock import MagicMock, patch from your_module import require_auth def test_require_auth_valid_token(): # 创建 Mock Redis 客户端 mock_redis = MagicMock() mock_redis.get.return_value = b"u1001" # 模拟返回 user_id # 构建一个模拟的 Request 对象 class MockRequest: def __init__(self): self.headers = {"Authorization": "Bearer abc123"} mock_request = MockRequest() # 定义一个被装饰的测试函数 @require_auth(redis_client=mock_redis) def test_func(request, user_id=None): return {"user_id": user_id} # 调用 result = test_func(mock_request) # 断言 assert result["user_id"] == "u1001" mock_redis.get.assert_called_once_with("auth:abc123") mock_redis.expire.assert_called_once_with("auth:abc123", 3600) def test_require_auth_invalid_header(): mock_redis = MagicMock() mock_request = type('MockRequest', (), {'headers': {}})() # 空 headers @require_auth(redis_client=mock_redis) def test_func(request): pass with pytest.raises(PermissionError, match="Invalid or missing Authorization header"): test_func(mock_request)这个测试覆盖了两个核心路径:成功和失败。关键技巧是:
- Mock 外部服务:
MagicMock替代redis.Redis,用return_value控制返回结果。 - 构造最小化测试对象:
MockRequest只实现headers属性,不引入任何框架依赖。 - 断言副作用:不仅检查返回值,还用
assert_called_once_with验证redis_client是否被正确调用,确保装饰器逻辑无误。
注意:不要试图测试
@require_auth(...)语法本身。那是 Python 解释器的工作,你只需测试require_auth(...)(func)返回的wrapper行为。这是单元测试的黄金法则——只测你写的代码,不测语言特性。
3.3 性能考量:装饰器的开销到底有多大?
有人担心装饰器会拖慢程序。答案是:绝大多数情况下,开销可以忽略不计,但某些设计会带来显著性能损失。我们来量化分析。
首先,装饰器本身的“包装”动作只在模块导入时执行一次,不是每次函数调用都执行。比如:
@timer def slow_function(): time.sleep(1)@timer这行代码,在slow_function定义时就被执行了,它返回wrapper并将slow_function重新绑定到wrapper。之后每次调用slow_function(),实际执行的是wrapper(),其开销就是wrapper函数体内的代码。
wrapper的典型开销包括:
- 函数调用开销:一次额外的函数调用(
wrapper→func),现代 Python 解释器对此优化很好,约 10-50ns。 *args, **kwargs解包:约 20-100ns,取决于参数数量。- 额外逻辑:比如
time.time()调用约 100ns,Redis 网络请求则是毫秒级。
所以,纯 Python 逻辑的装饰器(日志、计时、参数校验)开销在纳秒级,对整体性能无影响。真正的瓶颈永远是装饰器里调用的外部服务(DB、HTTP、Redis)。
但有一种写法会意外引入大开销:在装饰器定义时(而非wrapper内)执行重量级操作。例如:
# ❌ 危险:在 decorator 层就加载大文件 def load_config_and_decorate(): config = json.load(open("huge_config.json")) # 每次装饰都读一次! def decorator(func): @wraps(func) def wrapper(*args, **kwargs): # 使用 config... return func(*args, **kwargs) return wrapper return decorator如果这个装饰器被用在 100 个函数上,huge_config.json就会被读取 100 次!正确做法是把重量级操作移到wrapper内,并加缓存:
# ✅ 正确:延迟加载 + 缓存 def load_config_and_decorate(): _config_cache = None def decorator(func): @wraps(func) def wrapper(*args, **kwargs): nonlocal _config_cache if _config_cache is None: _config_cache = json.load(open("huge_config.json")) # 使用 _config_cache... return func(*args, **kwargs) return wrapper return decorator总结性能原则:装饰器的“准备”阶段(定义时)应极轻量;所有实际工作都应在wrapper的“执行”阶段完成,并根据需要缓存。把它当成一个懒加载的工厂,而不是一个急切执行的初始化器。
4. 常见问题与排查技巧实录
4.1 “TypeError: 'NoneType' object is not callable” —— 最经典的装饰器死亡报错
这个错误我至少见过 20 次,每次都让我心头一紧。它几乎总是由同一个原因引起:装饰器函数没有返回值,或返回了None。
回忆一下装饰器的契约:它必须返回一个可调用对象(通常是函数)。如果忘了return,Python 函数默认返回None,于是greet = add_logging(greet)这行就把greet变成了None,后续调用greet("Alice")就报错。
复现代码:
def broken_timer(func): def wrapper(*args, **kwargs): start = time.time() result = func(*args, **kwargs) end = time.time() print(f"{func.__name__} took {end - start:.4f}s") return result # ❌ 忘了 return wrapper!调试技巧:
第一反应看装饰器最后一行:是不是少写了
return wrapper?这是 80% 的情况。用
print检查返回值:在装饰器调用后立刻打印:decorated = broken_timer(greet) print(type(decorated), decorated) # 会输出 <class 'NoneType'> None启用 IDE 的语法检查:PyCharm 会在没有
return的函数末尾标黄警告,开启它能提前拦截。
修复很简单:补上return wrapper。但更深层的教训是:写装饰器时,把return wrapper当作和def wrapper一样重要的固定模板,写完def wrapper就立刻敲return wrapper,形成条件反射。
4.2 装饰器顺序陷阱:@A和@B谁先执行?
当一个函数被多个装饰器修饰时,顺序至关重要。例如:
@timer @require_auth(redis_client=redis_client) def api_endpoint(): pass这等价于:
def api_endpoint(): pass api_endpoint = timer(require_auth(redis_client=redis_client)(api_endpoint))执行顺序是从下到上(@require_auth先执行,@timer后执行),但调用顺序是从上到下(timer的wrapper先执行,再调用require_auth的wrapper,最后调用原函数)。
这意味着:timer测量的是整个认证+业务逻辑的耗时,而如果你把顺序颠倒:
@require_auth(redis_client=redis_client) @timer def api_endpoint(): pass那么timer只测量业务逻辑耗时,认证逻辑被排除在外。哪个更合理?取决于你的监控目标。如果是 API 响应时间 SLA,应该包含认证;如果是纯业务性能分析,则应排除。
更危险的是状态冲突。比如一个装饰器修改了kwargs(如注入user_id),另一个装饰器也依赖kwargs,但顺序错了,可能导致user_id还没注入就被读取。
排查技巧:
- 画调用栈:在每个
wrapper开头加print(f"Entering {func.__name__} wrapper"),运行看输出顺序。 - 阅读等价代码:遇到疑惑,立刻手动展开
@A @B func为A(B(func)),一目了然。 - 文档化顺序:在装饰器 docstring 中明确写出它期望的上下游装饰器,比如
@require_auth的文档可以写:“本装饰器向kwargs注入user_id,请确保下游装饰器(如@cache)能正确读取。”
4.3 装饰器与async/await:同步装饰器无法装饰异步函数
这是 Python 3.7+ 的高频坑。如果你尝试:
@timer async def async_fetch_data(): await asyncio.sleep(1) return "data"运行时会报错:TypeError: object None can't be used in 'await' expression或类似。原因是timer的wrapper是一个同步函数,它调用func(*args, **kwargs)时,得到的是一个coroutine对象,而不是结果。wrapper试图return coroutine_object,但调用者(比如await async_fetch_data())期望的是await一个协程,而不是直接返回协程。
解决方案是写一个异步装饰器,它返回一个async def wrapper:
import asyncio from functools import wraps def async_timer(func): @wraps(func) async def wrapper(*args, **kwargs): start = asyncio.get_event_loop().time() result = await func(*args, **kwargs) # ✅ await 而不是直接调用 end = asyncio.get_event_loop().time() print(f"{func.__name__} took {end - start:.4f}s") return result return wrapper关键区别:
async def wrapper:声明这是一个协程函数。await func(*args, **kwargs):等待协程执行完成,获取结果。return result:返回实际结果,不是协程对象。
同样,@require_auth如果要支持异步函数,也需要重写为async def wrapper,并在内部await redis_client.get(...)。
提示:不要试图用一个装饰器同时支持同步和异步函数。它们的调用协议完全不同(
func()vsawait func())。正确的做法是提供两个独立的装饰器:timer和async_timer,让用户按需选用。混用只会增加复杂度和 bug。
4.4 装饰器调试终极技巧:breakpoint()和inspect
当装饰器逻辑复杂,print不够用时,你需要进入调试器。但breakpoint()放在wrapper里,每次调用都停,太烦。这时inspect模块就派上用场了。
inspect.currentframe()可以获取当前栈帧,inspect.getframeinfo可以获取调用位置:
import inspect def debug_wrapper(func): @wraps(func) def wrapper(*args, **kwargs): # 只在特定条件下断点,比如某个参数为特定值 if args and args[0] == "debug_mode": frame = inspect.currentframe() info = inspect.getframeinfo(frame) print(f"Debug hit at {info.filename}:{info.lineno}") breakpoint() # 进入 pdb return func(*args,