云计算百科
云计算领域专业知识百科平台

一次讲清 Python 装饰器:从闭包、wraps 到 FastAPI 与 Agent 工程实践

一次讲清 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

这段模板有五个关键点:

  • func 保存被装饰的原函数;
  • wrapper 接管新的调用入口;
  • *args, **kwargs 转发调用参数;
  • return result 保留原函数返回值;
  • @wraps(func) 复制必要元数据并建立 __wrapped__ 链。
  • 为什么必须返回原结果

    错误写法:

    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

    异常处理还应遵守两个原则:

  • 记录后通常要继续 raise,不要悄悄把异常变成 None;
  • 不要用通用装饰器把所有异常转成同一种业务结果,否则会丢失语义和堆栈信息。
  • 十、同步与异步装饰器不能混为一谈

    同步包装函数如果直接调用异步函数,得到的只是协程对象:

    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 装饰器,可以归纳为五层理解:

  • 语法层:@decorator 大致等价于 func = decorator(func);
  • 机制层:函数是一等对象,闭包保存原函数,名称绑定到装饰结果;
  • 工程层:*args、**kwargs、返回值、wraps 和 ParamSpec 共同保证透明包装;
  • 框架层:FastAPI 路由、pytest 标记和 Agent Tool 常通过装饰器完成注册或元数据收集;
  • 设计层:装饰器适合抽离横切能力,但不应隐藏核心业务、忽略异步边界或制造不可控的导入副作用。
  • 真正写好装饰器,不是会套三层函数模板,而是能准确说明它在何时执行、改变了什么契约、失败时如何表现,以及是否值得增加这一层间接性。

    如果这篇文章帮你理清了闭包、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
    赞(0)
    未经允许不得转载:网硕互联帮助中心 » 一次讲清 Python 装饰器:从闭包、wraps 到 FastAPI 与 Agent 工程实践
    分享到: 更多 (0)

    评论 抢沙发

    评论前必须登录!