一次讲清 Python 装饰器:从闭包、wraps 到 FastAPI 与 Agent 工程实践
第一次看到下面的代码时,很多人会觉得装饰器像一种魔法:
@timer
def create_order(order_id: str) –> dict[str, str]:
return {"order_id": order_id}
其实,@timer 只是更容易阅读的语法。装饰器真正依赖的是 Python 中几个普通能力:
- 函数是一等对象;
- 函数可以作为参数和返回值;
- 内部函数可以通过闭包保存外部变量;
- 名称可以重新绑定到另一个对象;
- 函数或类定义完成后,可以立刻交给另一个可调用对象处理。
装饰器最核心的等价关系是:
@decorator
def task() –> None:
pass
大致等价于:
def task() –> None:
pass
task = decorator(task)
理解这行等价式,装饰器就不再神秘。
一、为什么函数可以被装饰
1. 函数是一等对象
函数可以赋给变量:
def say_hello() –> None:
print("hello")
action = say_hello
action()
可以作为参数:
from collections.abc import Callable
def run(action: Callable[[], None]) –> None:
action()
也可以作为返回值:
def choose_action(debug: bool) –> Callable[[], None]:
def debug_action() –> None:
print("debug")
def normal_action() –> None:
print("normal")
return debug_action if debug else normal_action
装饰器只是把这三种能力组合起来:接收原函数,返回一个函数,再把原来的名字绑定到返回值上。
2. 装饰器返回的不一定是函数
Python 语言层面要求装饰器表达式的结果是可调用对象,它会接收被装饰的函数或类;装饰器最终可以返回任何对象,并将该对象绑定到原名称。
工程上通常返回:
- 一个包装函数;
- 原函数本身;
- 一个包装类;
- 原类本身;
- 其他可调用对象。
如果返回 None,原函数名就会被重新绑定为 None,这也是注册型装饰器中常见的错误。
二、最小装饰器与闭包
def announce(func):
def wrapper():
print("before")
result = func()
print("after")
return result
return wrapper
使用:
@announce
def greet() –> str:
print("hello")
return "done"
装饰完成后的关系可以理解为:
greet 名称
↓
wrapper 函数
↓ 闭包保存
原始 greet 函数
虽然外层 announce() 已经执行结束,返回的 wrapper 仍然引用参数 func,因此原函数对象不会消失。这种“内部函数保存外部作用域变量”的机制就是闭包。
三、生产装饰器的基础模板
真实函数拥有各种位置参数、关键字参数和返回值,因此包装函数通常需要转发所有参数:
from functools import wraps
def log_call(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"calling {func.__name__}")
result = func(*args, **kwargs)
return result
return wrapper
这段模板有五个关键点:
为什么必须返回原结果
错误写法:
def broken(func):
def wrapper(*args, **kwargs):
func(*args, **kwargs)
return wrapper
无论原函数返回什么,装饰后的函数都会返回 None。除非装饰器明确要改变契约,否则应该原样返回结果。
四、functools.wraps 到底保留什么
如果不使用 wraps:
def simple(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
被装饰函数的 __name__、__doc__ 等信息可能表现为包装函数的信息。这会影响:
- 日志与错误定位;
- 自动生成文档;
- inspect 等反射工具;
- 测试与调试;
- 一些依赖函数签名和元数据的框架。
标准写法:
from functools import wraps
def simple(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
wraps 本质上是 update_wrapper 的便捷形式。它会更新包装函数的名称、限定名、文档、注解等常用属性,并设置 __wrapped__ 指向原函数。
可以通过以下方式访问最内层原函数:
from inspect import unwrap
original = unwrap(decorated_function)
需要注意:wraps 保留的是运行时元数据,不会自动让静态类型检查器理解包装前后的参数关系。类型安全还需要 ParamSpec 和 TypeVar。
五、使用 ParamSpec 写类型安全的装饰器
兼容 Python 3.10+ 的标准写法:
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
def traced(func: Callable[P, R]) –> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) –> R:
print(f"calling {func.__name__}")
return func(*args, **kwargs)
return wrapper
这里:
- P 捕获原函数完整的参数规格;
- P.args 表示位置参数;
- P.kwargs 表示关键字参数;
- R 保存返回值类型;
- 装饰前后的 Callable[P, R] 保持同一签名。
因此 IDE 和类型检查器仍然知道:
@traced
def add(a: int, b: int) –> int:
return a + b
add 仍然接收两个 int 并返回 int。
可以这样区分:
wraps
保护运行时元数据和 __wrapped__ 链
ParamSpec + TypeVar
保护静态参数与返回值关系
生产级装饰器通常两者都需要。
六、装饰器什么时候执行
装饰器表达式在函数定义时求值,而不是等到第一次调用函数时才求值。
def register(func):
print("decorate:", func.__name__)
return func
@register
def task() –> None:
print("run task")
模块导入并执行函数定义时,会先输出:
decorate: task
只有调用 task() 时,函数体才输出:
run task
可以分成两个阶段:
定义阶段 / 通常也是模块导入阶段
计算装饰器表达式
创建函数对象
调用装饰器
将返回值绑定到函数名
调用阶段
调用装饰后的对象
执行 wrapper 或原函数
因此不要在装饰器外层随意连接数据库、发送网络请求或加载大型模型,否则一次普通导入就可能触发昂贵副作用。
七、带参数装饰器为什么有三层
希望这样使用:
@retry(times=3)
def request_data() –> str:
...
Python 会先执行 retry(times=3),它必须返回真正接收 func 的装饰器,因此需要三层:
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
def retry(
*,
times: int,
exceptions: tuple[type[Exception], ...] = (TimeoutError,),
) –> Callable[[Callable[P, R]], Callable[P, R]]:
if times < 1:
raise ValueError("times must be at least 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, times + 1):
try:
return func(*args, **kwargs)
except exceptions:
if attempt == times:
raise
raise RuntimeError("unreachable")
return wrapper
return decorator
三层分别接收:
retry(…) 装饰器配置参数
decorator(func) 被装饰的原函数
wrapper(…) 调用原函数时的参数
重试不能只看语法,还要考虑业务语义:
- 只捕获明确的临时异常,不要默认吞掉所有 Exception;
- 有副作用的操作必须具备幂等性;
- 生产环境通常需要退避、抖动、超时和可观测性;
- 不要在同步装饰器里用阻塞休眠包装异步函数。
八、多个装饰器的顺序
@outer
@inner
def task() –> None:
pass
大致等价于:
task = outer(inner(task))
因此:
- 装饰阶段:从下往上应用,先 inner,再 outer;
- 调用进入:先进入最外层 outer,再进入 inner;
- 调用退出:先离开 inner,最后离开 outer。
示例:
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 before
inner before
原函数
inner after
outer after
装饰器顺序可能改变权限、缓存、事务和异常处理的语义,不能只为了排版随意调整。
九、异常安全:后置逻辑放在哪里
下面的计时逻辑在原函数抛异常时不会执行:
result = func(*args, **kwargs)
print("finished")
return result
若无论成功失败都要记录耗时,应使用 try/finally:
from functools import wraps
from time import perf_counter
def measure_time(func):
@wraps(func)
def wrapper(*args, **kwargs):
started = perf_counter()
try:
return func(*args, **kwargs)
finally:
elapsed = perf_counter() – started
print(f"{func.__name__}: {elapsed:.3f}s")
return wrapper
异常处理还应遵守两个原则:
十、同步与异步装饰器不能混为一谈
同步包装函数如果直接调用异步函数,得到的只是协程对象:
result = async_func(*args, **kwargs)
此时异步函数体还没有完整执行。同步计时、异常捕获和事务边界可能只覆盖“创建协程”,而没有覆盖真正的 await。
异步装饰器应使用 async def 并 await 原函数:
from collections.abc import Awaitable, Callable
from functools import wraps
from time import perf_counter
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
def measure_async(
func: Callable[P, Awaitable[R]],
) –> Callable[P, Awaitable[R]]:
@wraps(func)
async def wrapper(*args: P.args, **kwargs: P.kwargs) –> R:
started = perf_counter()
try:
return await func(*args, **kwargs)
finally:
elapsed = perf_counter() – started
print(f"{func.__name__}: {elapsed:.3f}s")
return wrapper
使用:
@measure_async
async def fetch_order(order_id: str) –> dict[str, str]:
return {"order_id": order_id}
如果一个装饰器同时支持同步和异步函数,通常需要在装饰阶段判断函数种类并返回对应包装器,同时用 overload 和必要的类型辅助保持签名。很多项目中分成 @measure_sync 与 @measure_async 反而更清楚。
十一、包装型与注册型装饰器
1. 包装型:改变调用行为
@measure_time
def build_report() –> str:
...
它通常返回 wrapper,用于日志、计时、权限、缓存、重试和事务等横切能力。
2. 注册型:收集函数信息
from collections.abc import Callable
from typing import Any
Tool = Callable[..., Any]
tool_registry: dict[str, Tool] = {}
def tool(name: str):
def decorator(func: Tool) –> Tool:
if name in tool_registry:
raise ValueError(f"duplicate tool: {name}")
tool_registry[name] = func
return func
return decorator
@tool("get_order")
def get_order(order_id: str) –> dict[str, str]:
return {"order_id": order_id, "status": "paid"}
注册型装饰器通常不改变调用行为,而是把函数登记到路由表、命令表、事件表或工具表,因此一般返回原函数。
还要注意:
- 注册发生在定义或导入阶段;
- 对应模块没有被导入,注册代码就不会执行;
- 同名注册应明确报错或规定覆盖策略;
- 进程内字典不是跨 Worker、跨容器的全局注册中心;
- 自动扫描模块容易引入循环导入和隐式副作用。
十二、FastAPI 路由与 Agent Tool
FastAPI 的路径操作装饰器:
from fastapi import FastAPI
app = FastAPI()
@app.get("/users/{user_id}")
def get_user(user_id: int) –> dict[str, int]:
return {"user_id": user_id}
它告诉 FastAPI:该函数负责处理指定路径的 GET 请求。路由装饰器还可以接收状态码、标签、响应模型等配置,并将相关信息用于 OpenAPI。
Agent Tool 装饰器的思想类似:
@tool("get_refund")
def get_refund(refund_id: str) –> dict[str, str]:
return {"refund_id": refund_id, "status": "pending"}
不过真实 Agent 框架通常还会读取:
- 函数名称和文档字符串;
- 参数类型注解;
- Pydantic 或 JSON Schema;
- 同步或异步调用方式;
- 权限、超时和可观测性信息。
这也是为什么自定义包装器应正确使用 wraps,并谨慎改变函数签名。
十三、常见内置与标准库装饰器
1. @property
把无参数方法暴露为属性式访问:
class Rectangle:
def __init__(self, width: float, height: float):
self.width = width
self.height = height
@property
def area(self) –> float:
return self.width * self.height
调用:
rectangle.area
属性不适合隐藏昂贵网络请求或明显有副作用的操作,否则调用方会误判成本。
2. @classmethod
类方法接收 cls,常用于替代构造器:
from typing import Self
class User:
def __init__(self, name: str):
self.name = name
@classmethod
def from_dict(cls, data: dict[str, str]) –> Self:
return cls(name=data["name"])
使用 cls(…) 而不是写死 User(…),可以让子类调用时创建子类实例。Self 需要 Python 3.11+;旧版本可使用 typing_extensions.Self 或绑定的 TypeVar。
3. @staticmethod
静态方法不会自动接收实例或类:
class User:
@staticmethod
def is_valid_name(name: str) –> bool:
return bool(name.strip())
如果函数和类的概念关系不强,放在模块级通常更直接。
4. @dataclass
from dataclasses import dataclass
@dataclass
class Point:
x: int
y: int
dataclass 是类装饰器,会根据类中声明的字段生成 __init__、__repr__、比较方法等可选能力。
5. @lru_cache
from functools import lru_cache
@lru_cache(maxsize=128)
def calculate_score(user_id: str) –> int:
return expensive_calculation(user_id)
使用缓存前必须确认:
- 参数可哈希;
- 相同参数可以复用旧结果;
- 函数没有不应跳过的副作用;
- 可以接受缓存失效与内存占用;
- 不要直接缓存普通异步函数的协程对象;
- 缓存是当前进程内的,不会自动跨 Worker 共享。
lru_cache 内部缓存结构具备线程安全性,但并不保证并发首次调用时原函数只执行一次。
十四、类装饰器
类也可以被装饰:
class_registry: dict[str, type] = {}
def register_model(name: str):
def decorator(cls: type) –> type:
if name in class_registry:
raise ValueError(f"duplicate model: {name}")
class_registry[name] = cls
return cls
return decorator
@register_model("user")
class UserModel:
pass
类装饰器可用于:
- 注册类;
- 自动补充类属性或方法;
- 将类替换为增强后的类;
- 生成数据模型能力。
修改类结构会影响继承、反射、序列化和类型检查。简单需求优先返回原类,复杂元编程应提供充分测试和文档。
十五、Python 装饰器与 Decorator Pattern
Python 的 @decorator 是语言语法,经典设计模式中的 Decorator Pattern 则通常通过对象组合增强对象能力。
class Coffee:
def cost(self) –> int:
return 10
class MilkDecorator:
def __init__(self, coffee: Coffee):
self.coffee = coffee
def cost(self) –> int:
return self.coffee.cost() + 2
两者共同的设计动机是:
不直接修改原始实现,通过包装或组合增加能力。
但不要把“Python 装饰器语法”和“面向对象装饰器模式”认为是完全相同的实现机制。
装饰器也常被视为 Python 中实现部分 AOP(面向切面编程)思想的轻量工具,因为日志、权限、监控和事务等逻辑会横跨多个业务函数。
十六、哪些场景适合装饰器
装饰器适合同时满足这些特点的逻辑:
- 多个函数或类都会使用;
- 不属于核心业务步骤;
- 能清楚描述为调用前、调用后或调用周围的增强;
- 可以独立测试和复用;
- 声明式写法比手动调用更清楚;
- 框架需要收集函数或类信息。
典型场景:
| 日志、指标、Tracing | 不记录密码、Token 等敏感信息 |
| 权限 | 通用入口权限适合,复杂业务策略应保持显式 |
| 缓存 | 处理失效、一致性、内存和多进程边界 |
| 重试 | 仅重试临时错误,有副作用操作必须幂等 |
| 事务 | 明确提交、回滚、嵌套和异步支持 |
| 注册 | 控制导入时机、重复名称和发现机制 |
| 限流 | 明确实例范围和分布式一致性 |
十七、常见反模式
1. 把核心业务藏进装饰器
@calculate_compensation
@auto_approve_refund
def handle_refund() –> None:
...
如果退款规则全部隐藏在装饰器中,阅读者很难看清业务流程。核心决策应放在显式 Service、Policy 或 Workflow 中。
2. 叠加太多装饰器
@a
@b
@c
@d
@e
def process() –> None:
...
层数过多会让执行顺序、异常归属、性能和返回值变化难以追踪。
3. 名字无法表达副作用
@magic、@enhance 无法告诉读者发生了什么。优先使用 @require_role、@measure_time、@retry 等明确名称。
4. 吞掉异常
try:
return func(*args, **kwargs)
except Exception:
return None
这种写法会把真正故障伪装成正常空值。
5. 同步包装异步
没有 await 的同步 wrapper 无法正确覆盖异步函数执行过程。
6. 导入阶段做重操作
装饰器外层的网络连接、模型加载或数据库访问会在模块导入时发生,使启动、测试和脚本执行变得不可预测。
7. 用装饰器改变公开签名却不说明
如果装饰器增加参数、删除参数或改变返回类型,应准确标注并写文档;不要伪装成“完全透明”的包装器。
十八、如何测试装饰器
装饰器本身也应有单元测试:
def test_traced_preserves_result() –> None:
@traced
def add(a: int, b: int) –> int:
return a + b
assert add(1, 2) == 3
还应检查:
- 位置参数和关键字参数是否正确转发;
- 返回值是否保留;
- 异常是否保持或按契约转换;
- 成功和失败路径的后置逻辑是否执行;
- __name__、__doc__ 和 __wrapped__ 是否保留;
- 多层装饰顺序是否符合预期;
- 异步函数是否真正被 await;
- 注册冲突和重复导入如何处理;
- 线程、进程和 Worker 边界是否符合设计。
借助 __wrapped__,测试有时可以绕开包装层直接验证原函数:
result = decorated_function.__wrapped__(...)
但这属于测试或调试技巧,业务代码通常不应绕过装饰器建立的权限、事务或校验边界。
十九、三套可复用模板
1. 普通同步包装器
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar
P = ParamSpec("P")
R = TypeVar("R")
def decorator(func: Callable[P, R]) –> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) –> R:
return func(*args, **kwargs)
return wrapper
2. 带配置参数的装饰器
def configured(
option: str,
) –> Callable[[Callable[P, R]], Callable[P, R]]:
def decorator(func: Callable[P, R]) –> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) –> R:
print(option)
return func(*args, **kwargs)
return wrapper
return decorator
3. 注册型装饰器
F = TypeVar("F", bound=Callable[..., object])
registry: dict[str, Callable[..., object]] = {}
def register(name: str) –> Callable[[F], F]:
def decorator(func: F) –> F:
if name in registry:
raise ValueError(f"duplicate registration: {name}")
registry[name] = func
return func
return decorator
模板只是起点。缓存、重试、事务和权限装饰器还必须根据业务补充生命周期、并发、幂等和安全设计。
二十、生产检查清单
写完装饰器后,逐项确认:
- 是否真的属于可复用的横切能力或注册逻辑;
- 是否使用 wraps 保留运行时元数据;
- 是否通过 ParamSpec、TypeVar 保留静态签名;
- 参数和返回值是否完整转发;
- 异常路径是否正确,是否误吞异常;
- 后置逻辑是否需要 finally;
- 是否分别支持同步与异步;
- 多个装饰器的顺序是否经过验证;
- 定义或导入阶段是否存在重副作用;
- 缓存、重试和事务是否符合业务语义;
- 是否考虑线程、进程、Worker 和容器边界;
- 是否有针对包装行为的单元测试。
总结
掌握 Python 装饰器,可以归纳为五层理解:
真正写好装饰器,不是会套三层函数模板,而是能准确说明它在何时执行、改变了什么契约、失败时如何表现,以及是否值得增加这一层间接性。
如果这篇文章帮你理清了闭包、wrapper、wraps、带参数装饰器和多层执行顺序,别忘了点个赞、收藏一下,方便以后写 Python、FastAPI 或 Agent 项目时随时回来查阅。
如果还有没看懂的地方,或者你想看异步装饰器、重试与缓存、FastAPI 权限装饰器或 Agent Tool 注册的完整实战,欢迎在评论区告诉我。后续还会继续分享 Python、软件设计、AI Agent 和后端工程相关内容,感兴趣的话点个关注,我们下一篇见!
参考资料
- Python 语言参考:Function Definitions 与 Decorators
- Python 标准库:functools.wraps 与 update_wrapper
- Python 标准库:functools.lru_cache
- Python 标准库:typing.ParamSpec
- Python 标准库:dataclasses
- Python 内置函数:property、classmethod 与 staticmethod
- FastAPI 官方教程:First Steps
- FastAPI 官方教程:Path Operation Configuration
网硕互联帮助中心



评论前必须登录!
注册