Python 基础体系 · 第 14/112 篇。示例统一以 Python 3.14 为语言基线;第三方库使用与其兼容的现代稳定版本,版本敏感行为会单独说明。
Python 装饰器:函数包装、参数化、类装饰器与元数据
装饰器(decorator)是 Python 中一种“在不直接修改目标定义主体的情况下,替换目标对象”的语法和编程模式。它经常用于日志、鉴权、缓存、重试、事务、路由注册、计时和运行时标记等场景。
但装饰器并不是函数调用前后自动执行的一层“魔法”。它首先是一条明确的赋值规则:
@decorator
def function(...):
...
大致等价于:
def function(...):
...
function = decorator(function)
装饰器表达式在函数定义时求值,装饰器返回的对象重新绑定到原来的函数名。多个装饰器则按照“从下到上应用、从外到内调用”的方式组合。Python 语言参考明确规定,下面的代码:
@f1(arg)
@f2
def func():
pass
等价于:
def func():
pass
func = f1(arg)(f2(func))
原始函数在装饰过程中不会临时绑定到 func 名称。(docs.python.org)
一、装饰器的最小模型:接收一个对象,返回一个对象
1. 装饰器的形式化条件
设目标对象为 ,装饰器为 。
一个合法的装饰器至少满足:
其中:
- 是被装饰对象;
- 是一个可调用对象;
- 是装饰表达式最终返回的对象;
- 会绑定到原来保存 的名称上。
对于函数装饰器,通常有:
但 Python 并不要求返回值必须是函数。只要返回对象满足后续使用场景,它也可以是:
- 一个可调用实例;
- 一个类;
- 一个描述符;
- 一个普通对象;
None或其他不可调用对象。
最后一种情况语法上可能成立,但通常会导致后续调用失败。
def replace_with_number(func):
return 42
@replace_with_number
def answer():
return "original"
print(answer)
# 42
# answer()
# TypeError: 'int' object is not callable
函数主体甚至没有被执行。装饰器发生在函数定义阶段,而不是调用阶段。
2. 装饰器调用发生在什么时候
看下面的代码:
def announce(func):
print(f"装饰 {func.__name__}")
return func
print("开始定义")
@announce
def work():
print("执行 work")
print("定义完成")
work()
输出为:
开始定义
装饰 work
定义完成
执行 work
这说明有两个不同时间点:
- 执行
def work时,创建函数对象并调用announce(work); - 执行
work()时,才执行函数主体。
装饰器中的初始化逻辑属于定义时逻辑,包装函数中的逻辑属于调用时逻辑。混淆这两个阶段,是理解装饰器生命周期时最常见的错误之一。
二、函数包装:把原函数保存到闭包中
最典型的装饰器会创建一个新的包装函数(wrapper),并在包装函数内部调用原函数。
from functools import wraps
def log_call(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"调用 {func.__name__}")
result = func(*args, **kwargs)
print(f"返回 {result!r}")
return result
return wrapper
@log_call
def add(a, b):
"""计算两个数的和。"""
return a + b
print(add(2, 3))
输出:
调用 add
返回 5
5
这里发生了以下绑定:
原始 add 函数对象
│
▼
log_call(add)
│
▼
wrapper 函数对象
│
▼
名称 add 现在指向 wrapper
调用 add(2, 3) 时,实际调用的是 wrapper(2, 3)。wrapper 再通过闭包变量 func 调用原始函数。
1. 为什么需要 *args 和 **kwargs
如果包装器写死参数:
def bad_log(func):
def wrapper(a, b):
return func(a, b)
return wrapper
那么它只能包装恰好接收两个位置参数的函数:
@bad_log
def multiply(a, b):
return a * b
下面这些情况就可能失败:
@bad_log
def greet(name, *, punctuation="!"):
return f"Hello, {name}{punctuation}"
因为包装器的签名不支持关键字参数 punctuation。
更通用的形式是:
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
其中:
args保存位置参数元组;kwargs保存关键字参数字典;func(*args, **kwargs)将参数重新转发给原函数。
但“能转发”不等于“接口完全等价”。包装函数的真实 Python 签名仍然是:
(*args, **kwargs)
这会影响文档生成、IDE 提示、运行时检查和依赖签名的框架。
三、参数绑定、位置参数和关键字参数
装饰器如果需要检查或修改参数,不能只把参数当作一个简单的元组和字典处理。
例如:
def show_arguments(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("位置参数:", args)
print("关键字参数:", kwargs)
return func(*args, **kwargs)
return wrapper
@show_arguments
def example(a, b=10, *, debug=False):
return a + b
调用:
example(1, debug=True)
包装器看到的是:
args == (1,)
kwargs == {"debug": True}
但它还不知道:
1对应参数a;b使用默认值10;debug是关键字专用参数。
如果装饰器需要按参数名称进行逻辑处理,应使用 inspect.signature():
import inspect
from functools import wraps
def require_positive(func):
signature = inspect.signature(func)
@wraps(func)
def wrapper(*args, **kwargs):
bound = signature.bind(*args, **kwargs)
bound.apply_defaults()
for name, value in bound.arguments.items():
if name in {"amount", "limit"} and value <= 0:
raise ValueError(f"{name} 必须为正数")
return func(*args, **kwargs)
return wrapper
@require_positive
def transfer(amount, limit=100):
return min(amount, limit)
print(transfer(20))
# 20
# transfer(-1)
# ValueError: amount 必须为正数
Signature.bind() 根据函数签名执行参数绑定,apply_defaults() 则将未显式传入的默认参数补入绑定结果。inspect.signature() 默认会沿着 __wrapped__ 链进行解包,这也是 functools.wraps() 能帮助工具恢复原始签名信息的原因之一。(docs.python.org)
位置专用参数与关键字专用参数
Python 函数可能包含:
def request(method, url, /, *, timeout=5):
...
这里:
/之前的method和url只能按位置传递;*之后的timeout只能按关键字传递。
装饰器不能随意把这些参数改成另一种调用方式,否则可能破坏 API 契约。
例如,下面的调用是合法的:
request("GET", "/users", timeout=2)
但下面的调用不合法:
request(method="GET", url="/users")
# TypeError
因此,使用 *args, **kwargs 转发通常比手动重建调用更安全。若确实要修改参数,应先通过 Signature.bind() 得到语义明确的绑定结果,再重新组织调用。
四、闭包是函数装饰器的核心支撑
函数包装器通常依赖闭包保存原函数:
def decorate(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
wrapper 使用了外层函数 decorate 的局部变量 func。即使 decorate 已经返回,func 仍然会被包装器保留。
可以通过 __closure__ 观察这一点:
def decorate(func):
def wrapper():
return func()
return wrapper
def original():
return "ok"
wrapped = decorate(original)
print(wrapped.__closure__)
print(wrapped.__closure__[0].cell_contents)
# <function original ...>
闭包中的变量不是简单复制出来的普通值,而是保存在 cell 中的外部绑定。多个内部函数还可能共享同一个 cell,因此 nonlocal 可以修改闭包状态。
有状态装饰器
from functools import wraps
def count_calls(func):
count = 0
@wraps(func)
def wrapper(*args, **kwargs):
nonlocal count
count += 1
print(f"{func.__name__} 已调用 {count} 次")
return func(*args, **kwargs)
return wrapper
@count_calls
def ping():
return "pong"
ping()
ping()
输出:
ping 已调用 1 次
ping 已调用 2 次
这里 count 位于 count_calls 的局部作用域中,但由 wrapper 通过闭包访问。若没有 nonlocal count,执行 count += 1 会被解释为对局部变量的赋值,进而产生:
UnboundLocalError
原因是 count += 1 等价于读取旧值后再赋值;一旦函数体内出现赋值,名称默认属于当前局部作用域。
闭包的迟绑定问题
装饰器工厂批量创建包装器时,尤其容易遇到迟绑定:
def make_decorators(names):
decorators = []
for name in names:
def decorator(func):
def wrapper():
return f"{name}: {func()}"
return wrapper
decorators.append(decorator)
return decorators
这些 decorator 函数引用的 name 是同一个循环作用域中的变量。真正调用包装器时,循环已经结束,因此它们可能都读取最后一个 name。
修复方式是把当前值绑定为默认参数:
def make_decorators(names):
decorators = []
for name in names:
def decorator(func, prefix=name):
@wraps(func)
def wrapper():
return f"{prefix}: {func()}"
return wrapper
decorators.append(decorator)
return decorators
或者使用参数化装饰器工厂,让当前值通过函数调用形成独立闭包:
def label(prefix):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return f"{prefix}: {func(*args, **kwargs)}"
return wrapper
return decorator
装饰器与闭包的关系可以概括为:
装饰器工厂创建环境
│
▼
环境中保存配置和原函数
│
▼
包装函数通过闭包读取配置和原函数
│
▼
包装函数替换原名称
五、参数化装饰器:三层调用结构
无参数装饰器的调用结构是:
@decorate
def func():
...
等价于:
func = decorate(func)
参数化装饰器的结构则多一层:
@retry(times=3)
def fetch():
...
等价于:
func = retry(times=3)(func)
因此,retry(times=3) 本身并没有接收函数;它先返回一个真正接收函数的装饰器。
from functools import wraps
def retry(times):
if times < 1:
raise ValueError("times 必须至少为 1")
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except Exception as exc:
last_error = exc
print(f"第 {attempt} 次失败: {exc}")
raise last_error
return wrapper
return decorator
使用:
attempts = 0
@retry(times=3)
def unstable():
global attempts
attempts += 1
if attempts < 3:
raise RuntimeError("暂时失败")
return "成功"
print(unstable())
输出:
第 1 次失败: 暂时失败
第 2 次失败: 暂时失败
成功
这里有三个不同对象:
| 名称 | 含义 |
|---|---|
retry |
装饰器工厂 |
retry(times=3) 的返回值 |
真正的装饰器 |
retry(times=3)(unstable) 的返回值 |
包装后的函数 |
参数化装饰器的状态边界
配置通常在定义阶段固定:
@retry(times=3)
def unstable():
...
times=3 被保存到装饰器闭包中,之后每次调用 unstable() 都使用这份配置。
如果需要每次调用都传入不同的重试次数,重试次数就不应设计成装饰器参数,而应设计成函数参数:
def retry_runtime(func):
@wraps(func)
def wrapper(*args, retries=3, **kwargs):
for attempt in range(retries):
try:
return func(*args, **kwargs)
except Exception:
if attempt == retries - 1:
raise
return wrapper
这两种设计的语义不同:
装饰器参数:定义一个具有固定策略的新函数
函数参数:每次调用时选择策略
同时支持 @decorate 和 @decorate(...)
有些装饰器希望两种写法都支持:
@trace
def f():
...
@trace(prefix="API")
def g():
...
可以使用一个可选参数区分:
from functools import wraps
def trace(func=None, *, prefix="CALL"):
def decorator(target):
@wraps(target)
def wrapper(*args, **kwargs):
print(f"{prefix}: {target.__name__}")
return target(*args, **kwargs)
return wrapper
if func is None:
return decorator
return decorator(func)
关键点在于:
@trace会执行trace(func);@trace(prefix="API")会先执行trace(func=None, prefix="API"),得到decorator;- 随后 Python 再执行
decorator(g)。
这类接口虽然灵活,但也增加了参数判别复杂度。如果装饰器本身不需要同时支持两种形式,优先选择一种明确写法。
六、多个装饰器的组合顺序
考虑:
def outer(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("outer before")
result = func(*args, **kwargs)
print("outer after")
return result
return wrapper
def inner(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("inner before")
result = func(*args, **kwargs)
print("inner after")
return result
return wrapper
@outer
@inner
def task():
print("task")
绑定过程是:
task = outer(inner(task))
调用过程是:
outer before
inner before
task
inner after
outer after
可以用嵌套函数表示:
outer.wrapper(
inner.wrapper(
task
)
)
这会直接影响:
- 日志顺序;
- 异常处理边界;
- 事务和锁的覆盖范围;
- 缓存命中时哪些逻辑会被跳过;
- 鉴权与限流的先后关系。
例如:
@cache
@authorize
def get_data():
...
等价于:
get_data = cache(authorize(get_data))
缓存包装器在外层,因此缓存命中时可能直接返回结果,内部的鉴权包装器不会执行。若安全策略要求每次调用都鉴权,就不能仅凭装饰器名称判断顺序,必须分析实际嵌套结构。
七、异常、返回值和控制流必须透明
一个行为透明的包装器通常需要做到三点:
- 返回原函数的返回值;
- 不无故吞掉异常;
- 在异常路径上正确执行清理逻辑。
from functools import wraps
def timing(func):
@wraps(func)
def wrapper(*args, **kwargs):
started = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
elapsed = time.perf_counter() - started
print(f"{func.__name__}: {elapsed:.6f}s")
return wrapper
这里使用 finally,因此无论原函数:
- 正常返回;
- 抛出异常;
- 在包装器中被中断;
计时逻辑都会执行。
完整示例:
import time
from functools import wraps
def timing(func):
@wraps(func)
def wrapper(*args, **kwargs):
started = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
elapsed = time.perf_counter() - started
print(f"{func.__name__}: {elapsed:.6f}s")
return wrapper
@timing
def divide(a, b):
return a / b
print(divide(10, 2))
try:
divide(10, 0)
except ZeroDivisionError:
print("捕获到除零异常")
输出类似:
divide: 0.000001s
5.0
divide: 0.000001s
捕获到除零异常
如果包装器写成:
def bad(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception:
return None
return wrapper
它改变了函数的异常契约:原本调用者可以通过异常发现失败,现在失败被转换成了 None。这不一定错误,但必须是明确的 API 设计,而不是为了“让程序继续运行”随手吞异常。
更危险的是:
def worse(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
finally:
return None
return wrapper
finally 中的 return 会覆盖 try 中的返回值,甚至覆盖异常传播,导致原函数的结果和异常都消失。Python 3.14 对 finally 中出现 return、break 或 continue 会发出 SyntaxWarning,因为这种控制流很容易隐藏异常。(docs.python.org)
八、functools.wraps:恢复包装函数的元数据
没有 wraps 时:
def decorate(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
@decorate
def calculate(x: int) -> int:
"""计算平方。"""
return x * x
print(calculate.__name__)
print(calculate.__doc__)
print(calculate.__annotations__)
典型结果是:
wrapper
None
{}
因为 calculate 这个名称现在绑定的是 wrapper,而 wrapper 的元数据来自包装器定义本身。
使用 @wraps(func):
from functools import wraps
def decorate(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
再观察:
@decorate
def calculate(x: int) -> int:
"""计算平方。"""
return x * x
print(calculate.__name__)
# calculate
print(calculate.__doc__)
# 计算平方。
print(calculate.__annotations__)
# {'x': <class 'int'>, 'return': <class 'int'>}
print(calculate.__wrapped__)
# 原始 calculate 函数对象
functools.wraps() 是 functools.update_wrapper() 的便捷形式。默认情况下,它会复制或更新包括 __module__、__name__、__qualname__、__annotations__、__type_params__ 和 __doc__ 在内的元数据,并把原函数放到包装器的 __wrapped__ 属性中。(docs.python.org)
这里的“复制”并不意味着创建一个完全独立的函数对象。它主要是让包装器在名称、文档、注解和内省行为上表现得像原函数。
__wrapped__ 的意义
__wrapped__ 是一个明确的包装链入口:
from inspect import unwrap
original = unwrap(calculate)
print(original.__name__)
# calculate
多个装饰器使用 wraps 时,会形成链:
最外层 wrapper
│ __wrapped__
▼
中间层 wrapper
│ __wrapped__
▼
原始函数
inspect.unwrap() 会沿着这个链继续查找,遇到循环则抛出 ValueError。inspect.signature() 默认也会跟随包装链,因此合理使用 wraps 能让工具看到原始函数签名。(docs.python.org)
如果需要查看包装器自身的真实签名,可以关闭解包:
import inspect
print(inspect.signature(calculate))
print(inspect.signature(calculate, follow_wrapped=False))
常见结果分别类似:
(x: int) -> int
(*args, **kwargs)
__signature__ 与显式签名
wraps 只能复制元数据,不能自动让包装器真正接受原函数的参数形式。它主要改善内省结果。
如果包装器改变了公开接口,可以显式设置 __signature__:
import inspect
from functools import wraps
def expose_original_signature(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
wrapper = wraps(func)(wrapper)
wrapper.__signature__ = inspect.signature(func)
return wrapper
不过 inspect 文档指出,CPython 对 __signature__ 的具体处理属于实现细节,不应把它误认为所有 Python 实现都必须完全相同的底层机制。(docs.python.org)
九、函数对象可以携带自定义元数据
用户定义的函数对象支持任意属性:
def endpoint(path, method):
def decorator(func):
func.route_path = path
func.route_method = method
return func
return decorator
@endpoint("/users", "GET")
def list_users():
return ["alice", "bob"]
print(list_users.route_path)
# /users
print(list_users.route_method)
# GET
函数属性适合表达“这个函数属于什么类别”或“应如何被注册”。框架可以扫描模块中的函数,读取这些属性并建立路由表:
def collect_routes(namespace):
routes = []
for obj in namespace.values():
if callable(obj) and hasattr(obj, "route_path"):
routes.append(
{
"path": obj.route_path,
"method": obj.route_method,
"handler": obj,
}
)
return routes
但是,元数据放在原函数还是包装函数上,需要特别注意。
元数据放置位置
from functools import wraps
def mark(func):
func.is_public = True
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
此时:
print(hasattr(marked, "is_public"))
不一定得到预期结果,因为属性先写在 func 上,而最终绑定到名称的是 wrapper。
更直接的写法是:
def mark(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
wrapper.is_public = True
return wrapper
如果希望两层都能读取,也可以同时设置,但这会引入同步问题。更稳妥的方式是约定元数据的唯一归属位置,并让扫描工具明确使用:
inspect.unwrap(obj)
或者检查包装对象自身。
十、装饰器与函数注解:Python 3.14 的版本边界
Python 3.14 中,注解默认采用延迟求值。也就是说,注解表达式通常不会在定义代码执行时立即求值,而是在需要读取和解析时再求值;annotationlib 提供了相应的评估工具。注解本身仍然不自动改变函数运行语义,只有使用注解的工具或框架才会赋予它额外含义。(docs.python.org)
这对装饰器有两个影响。
第一,装饰器不要默认假设:
func.__annotations__
中的每个值都已经是最终运行时类型对象。注解可能需要进一步评估,也可能包含会执行任意代码的表达式。因此,处理外部输入或不可信模块时,自动解析注解需要考虑安全风险。inspect.get_annotations() 的文档明确提醒,评估注解可能执行其中包含的任意代码。(docs.python.org)
第二,@wraps 会把注解相关属性复制到包装器,但这不等于装饰器自动理解了注解。例如:
from functools import wraps
def validate_return(func):
@wraps(func)
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
return result
return wrapper
这个装饰器只是保留注解,并没有验证返回值。若要验证,就必须明确调用注解解析工具并定义验证语义,不能把“保留元数据”和“执行类型检查”混为一谈。
十一、函数装饰器作用于方法时:描述符会参与绑定
类中的普通函数不是简单地存储后直接返回。函数对象作为类属性访问时,会通过描述符协议产生绑定方法。
class Greeter:
def greet(self, name):
return f"Hello, {name}"
obj = Greeter()
print(Greeter.__dict__["greet"])
# 原始函数对象
print(obj.greet)
# 绑定方法对象
绑定方法包含:
__self__:绑定的实例;__func__:原始函数对象。
Python 数据模型说明,用户定义函数是非数据描述符;从实例访问时,会把实例绑定为第一个参数。(docs.python.org)
因此,下面这个装饰器可以正常用于实例方法:
from functools import wraps
def log_method(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"调用 {func.__qualname__}")
return func(*args, **kwargs)
return wrapper
class UserService:
@log_method
def find(self, user_id):
return {"id": user_id}
service = UserService()
print(service.find(7))
调用路径是:
service.find(7)
│
▼
wrapper(service, 7)
│
▼
func(service, 7)
如果包装器错误地省略 self:
def bad_method_decorator(func):
def wrapper(user_id):
return func(user_id)
return wrapper
调用实例方法时会出现参数错位或参数数量错误,因为绑定机制仍然会把实例传给包装器。
装饰器顺序与 staticmethod、classmethod
staticmethod 和 classmethod 本身也参与方法绑定,因此顺序会影响结果。
推荐理解为“先应用靠近函数的装饰器”:
class Example:
@classmethod
@log_method
def make(cls, value):
return cls(value)
等价于:
Example.make = classmethod(log_method(make))
此时 log_method 包装的是普通函数,随后 classmethod 再把类绑定为第一个参数。
另一种顺序:
class Example:
@log_method
@classmethod
def make(cls, value):
return cls(value)
等价于:
Example.make = log_method(classmethod(make))
这时 log_method 接收的是 classmethod 对象,而不是普通函数。包装器是否能正确处理该对象,取决于具体实现。对于需要同时支持这些内置描述符的装饰器,必须明确测试类访问、实例访问和直接访问类字典的行为。
十二、类装饰器:替换类对象,而不是逐个装饰方法
类装饰器的语法与函数装饰器相同:
@decorate_class
class Service:
...
大致等价于:
class Service:
...
Service = decorate_class(Service)
类主体会先执行,类对象创建完成后,类装饰器才被调用。类装饰器拿到的是完整的类对象,因此可以:
- 添加类属性;
- 注册类;
- 修改或替换方法;
- 返回子类;
- 返回代理类;
- 返回完全不同的对象。
添加类元数据
def registered(cls):
cls.is_registered = True
return cls
@registered
class Plugin:
pass
print(Plugin.is_registered)
# True
类装饰器注册子类
registry = {}
def register(name):
def decorator(cls):
if name in registry:
raise ValueError(f"{name!r} 已注册")
registry[name] = cls
cls.plugin_name = name
return cls
return decorator
@register("json")
class JsonPlugin:
pass
print(registry["json"] is JsonPlugin)
# True
这个注册动作发生在模块执行阶段。导入模块就会执行类定义和注册逻辑,因此导入顺序、重复导入和循环导入都会成为注册系统的一部分状态。
返回新类
def add_id(cls):
original_init = cls.__init__
def __init__(self, *args, **kwargs):
self.id = None
original_init(self, *args, **kwargs)
cls.__init__ = __init__
return cls
这里修改的是原类对象本身。
也可以返回一个子类:
def subclass_with_flag(cls):
class Wrapped(cls):
enabled = True
return Wrapped
但这会改变身份关系:
@subclass_with_flag
class Service:
pass
print(Service.__name__)
# Wrapped
原始类名现在指向新类,__module__、__qualname__、序列化、类型比较和调试信息都可能受到影响。类装饰器返回新类时,不能只验证“实例能否正常运行”,还要检查类身份和元数据是否符合调用方预期。
类装饰器与元类的边界
类装饰器和元类都能影响类创建,但介入时机不同:
执行 class 主体
│
▼
元类创建类对象
│
▼
类装饰器接收已创建的类对象
│
▼
类名重新绑定
元类参与类对象的创建过程,类装饰器则处理创建完成后的结果。若需求是:
- 控制类创建协议;
- 改变继承解析;
- 拦截类命名空间;
- 统一影响整个继承体系;
元类或 __init_subclass__ 可能更合适。
若需求只是:
- 注册类;
- 添加标记;
- 修改少量属性;
类装饰器通常更直接。
十三、可调用类实例也可以作为装饰器
装饰器只要求对象可调用,因此类实例也可以使用:
from functools import update_wrapper
class CountCalls:
def __init__(self, func):
self.func = func
self.count = 0
update_wrapper(self, func)
def __call__(self, *args, **kwargs):
self.count += 1
return self.func(*args, **kwargs)
@CountCalls
def add(a, b):
return a + b
print(add(1, 2))
print(add.count)
# 3
# 1
这种实现的状态保存在实例属性中,而不是闭包变量中。
两种写法的状态模型不同:
闭包装饰器:
状态通常是外层局部变量,由 nonlocal 修改
可调用对象:
状态通常是 self 的属性,由实例方法修改
可调用对象适合状态较多、需要拆分多个辅助方法或希望显式暴露状态的场景。但它也会引入方法绑定问题。
例如,把 CountCalls 用于实例方法时:
class Service:
@CountCalls
def work(self, value):
return value * 2
类属性中的 work 现在是 CountCalls 实例。普通用户定义对象如果没有实现 __get__,不会自动像函数那样绑定 self。于是:
Service().work(3)
可能把 3 直接传给 CountCalls.__call__,而不是把实例作为第一个参数传入。
要让可调用装饰器实例支持方法绑定,需要实现描述符:
from functools import update_wrapper
class CountCalls:
def __init__(self, func):
self.func = func
self.count = 0
update_wrapper(self, func)
def __get__(self, instance, owner=None):
if instance is None:
return self
return self.__class__(
self.func.__get__(instance, owner)
)
def __call__(self, *args, **kwargs):
self.count += 1
return self.func(*args, **kwargs)
不过这个版本还涉及计数器是否应该由每个实例独立拥有的问题。若希望统计所有实例的总调用次数,就不能在 __get__ 中创建独立计数器;若希望每个实例独立统计,就需要进一步定义状态归属。
这正是函数装饰器、可调用对象和描述符之间的连接点:方法装饰不是单纯的函数替换,属性访问协议会参与最终调用行为。
十四、异步函数必须使用异步包装器
普通函数包装器不能直接代替异步控制流:
def bad_async_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
这个包装器返回的是协程对象,表面上可能仍能工作,但如果要在调用前后执行异步操作,普通 def 无法使用 await。
正确形式是:
import asyncio
from functools import wraps
def async_log(func):
@wraps(func)
async def wrapper(*args, **kwargs):
print("开始")
try:
return await func(*args, **kwargs)
finally:
print("结束")
return wrapper
@async_log
async def fetch():
await asyncio.sleep(0.01)
return "data"
print(asyncio.run(fetch()))
调用过程是:
fetch()
│
▼
创建 wrapper 协程对象
│
▼
事件循环运行 wrapper
│
▼
await 原始 func 返回的协程对象
asyncio 使用 async def 和 await 表达协程,并通过事件循环调度协程和任务。装饰器如果改变了协程函数的调用结构,就必须保证返回值仍然是可等待对象。(docs.python.org)
一个常见错误是:
def bad_async_log(func):
@wraps(func)
async def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
print(result)
return result
return wrapper
这里 result 是协程对象而不是最终结果。应写成:
result = await func(*args, **kwargs)
对于同时支持同步函数和异步函数的装饰器,不能只根据返回值判断,因为调用同步函数和调用异步函数的语义不同。通常需要在装饰阶段检查目标是否为协程函数,并分别创建同步包装器和异步包装器。
十五、装饰器中的可变状态与并发
有状态装饰器会长期持有状态,因此必须明确并发模型。
下面的计数器在单线程同步代码中通常可以正常工作:
def count_calls(func):
count = 0
@wraps(func)
def wrapper(*args, **kwargs):
nonlocal count
count += 1
return func(*args, **kwargs)
return wrapper
但如果多个线程同时调用,count += 1 并不应被当作抽象意义上的不可分割事务。它包含读取、计算和写回。若计数必须准确,需要使用锁:
import threading
from functools import wraps
def thread_safe_count(func):
lock = threading.Lock()
count = 0
@wraps(func)
def wrapper(*args, **kwargs):
nonlocal count
with lock:
count += 1
current = count
print(f"第 {current} 次调用")
return func(*args, **kwargs)
return wrapper
锁只保护计数状态,不应无条件包住整个业务函数:
with lock:
return func(*args, **kwargs)
否则所有调用会被串行化,甚至可能在业务函数再次调用被装饰函数时造成死锁或严重降低并发性。
异步代码中也不能直接用线程锁代替异步同步原语。若状态只属于当前异步任务,还需要考虑任务之间是否共享同一装饰器闭包。装饰器定义在模块级别时,闭包状态通常由所有调用者共享;如果业务语义要求“每个请求独立”,共享闭包状态就可能是错误的设计。
十六、常见失败表现与诊断方法
1. 忘记返回包装器
def broken(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
这里 broken(func) 隐式返回 None:
@broken
def f():
return 1
因此 f 会被重新绑定为 None。
2. 忘记返回原函数结果
def missing_return(func):
def wrapper(*args, **kwargs):
func(*args, **kwargs)
return wrapper
原函数虽然执行了,但包装器返回 None。
3. 忘记转发关键字参数
def missing_kwargs(func):
def wrapper(*args):
return func(*args)
return wrapper
当原函数依赖关键字参数时,调用会失败。
4. 忘记 wraps
表现通常包括:
func.__name__ == "wrapper"
func.__doc__ is None
inspect.signature(func) == (*args, **kwargs)
这可能进一步影响:
- API 文档;
- 路由框架;
- 依赖注入;
- 参数校验;
- 测试工具;
- 调试和追踪系统。
5. 包装器捕获了错误作用域
def factory():
prefix = "A"
def decorator(func):
def wrapper():
return prefix + func()
return wrapper
return decorator
这里 prefix 是闭包变量。如果之后代码通过 nonlocal 修改它,已经创建的所有包装器可能观察到新值。需要的是“定义时快照”还是“调用时读取”,必须从语义上决定。
6. 包装器破坏方法绑定
尤其需要检查以下组合:
@decorator
@staticmethod
def method():
...
@decorator
@classmethod
def method(cls):
...
@decorator
def method(self):
...
诊断时可以分别观察:
print(Class.__dict__["method"])
print(instance.method)
print(getattr(instance.method, "__self__", None))
print(getattr(instance.method, "__func__", None))
还可以检查包装链:
import inspect
print(inspect.unwrap(instance.method))
print(inspect.signature(instance.method))
print(inspect.signature(instance.method, follow_wrapped=False))
十七、一个完整的生产型示例:重试、日志和元数据
下面组合一个同步重试装饰器。它包含:
- 参数化配置;
wraps元数据保留;- 参数校验;
- 异常传播;
- 重试次数状态;
- 最后一次异常重新抛出。
from __future__ import annotations
from functools import wraps
from typing import Callable, TypeVar, ParamSpec
P = ParamSpec("P")
R = TypeVar("R")
def retry(
*,
attempts: int = 3,
exceptions: tuple[type[Exception], ...] = (Exception,),
) -> Callable[[Callable[P, R]], Callable[P, R]]:
if attempts < 1:
raise ValueError("attempts 必须至少为 1")
def decorator(func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
for attempt in range(1, attempts + 1):
try:
return func(*args, **kwargs)
except exceptions:
if attempt == attempts:
raise
print(
f"{func.__qualname__} 失败,"
f"准备第 {attempt + 1} 次尝试"
)
raise AssertionError("不可达代码")
wrapper.retry_attempts = attempts
return wrapper
return decorator
使用:
calls = 0
@retry(attempts=3, exceptions=(RuntimeError,))
def load_data() -> str:
"""加载数据。"""
global calls
calls += 1
if calls < 3:
raise RuntimeError("服务暂时不可用")
return "ready"
print(load_data())
print(load_data.__name__)
print(load_data.__doc__)
print(load_data.retry_attempts)
预期输出:
load_data 失败,准备第 2 次尝试
load_data 失败,准备第 3 次尝试
ready
load_data
加载数据。
3
这个例子有几个重要边界:
exceptions只捕获明确指定的异常,避免把KeyboardInterrupt、SystemExit等系统级控制流误当成业务失败;- 最后一次失败使用裸
raise,保留原始异常和 traceback; attempts在装饰器工厂执行时校验,因此模块定义阶段就能发现非法配置;wrapper.retry_attempts是包装器元数据,@wraps负责保留原函数的标准元数据;ParamSpec只帮助静态类型检查器表达参数转发关系,不会自动改变运行时包装器签名。
十八、选择函数包装、类装饰器还是描述符
三者解决的问题不同。
函数包装
适合:
调用前后增加逻辑
修改或检查参数
转换返回值
处理异常
记录指标
其核心是:
wrapper(*args, **kwargs) -> func(*args, **kwargs)
类装饰器
适合:
注册类
添加类级元数据
修改类属性
批量处理类定义
其核心是:
new_class = decorate_class(old_class)
描述符
适合:
改变属性访问和绑定行为
实现 property 风格的数据访问
控制实例属性读写
让可调用对象表现得像方法
其核心是:
attribute.__get__(instance, owner)
函数本身作为非数据描述符参与方法绑定,property 则是数据描述符;数据描述符通常优先于实例字典中的同名属性,普通方法则可以被实例属性覆盖。(docs.python.org)
因此,如果问题是“在函数调用前后加逻辑”,从函数装饰器开始;如果问题是“这个类应该被注册或标记”,考虑类装饰器;如果问题是“访问 obj.x 时应该执行什么协议”,就应直接理解描述符,而不是把所有行为都强行写成装饰器。
十九、装饰器的核心边界
装饰器的本质可以压缩为四个动作:
1. 在定义阶段求值装饰器表达式
2. 创建或取得原始对象
3. 调用装饰器得到替代对象
4. 用替代对象重新绑定名称
函数包装进一步增加了:
5. 通过闭包或对象属性保存原始函数
6. 在调用时转发参数
7. 返回结果或传播异常
8. 维护元数据和包装链
真正可靠的装饰器必须同时回答这些问题:
- 它是在定义阶段做什么,还是在调用阶段做什么?
- 它接收的是函数、方法描述符还是类?
- 它是否保持参数转发语义?
- 它是否保持返回值和异常契约?
- 它是否保留
__name__、__doc__、注解和__wrapped__? - 它是否改变了方法绑定?
- 它的状态由谁共享,是否需要同步?
- 多个装饰器叠加时,实际顺序是什么?
- 对同步函数和异步函数,包装器的控制流是否匹配?
- 类装饰器返回的是原类、修改后的原类,还是新类?
当这些问题都能被明确回答时,装饰器就不再是语法糖,而是一种可推导、可测试、可诊断的对象替换机制。
系列导航与关联阅读
- 系列入口:Python 完整学习路线:从语言模型、并发到 Web、数据、AI 与生产交付
- 上一篇:Python 作用域与闭包:LEGB、nonlocal、global 和迟绑定
- 下一篇:Python 可迭代对象与迭代器:协议、惰性、耗尽和组合
- 延伸:Python 描述符与 property:属性访问、绑定方法和 ORM 基础
官方资料
本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理;正文、示例与工程清单由 WR BLOG 编写。

评论
0 条讨论