九老板的工作室 — Jay's Studio

从一个 `@` 符号开始：它不是魔法

在 Python 项目中，装饰器几乎无处不在：它可以声明 Web 路由、缓存计算结果、准备测试夹具，也可以改变方法的绑定方式。常见写法包括：

text

@app.get("/users")       # FastAPI 路由
@cache                    # 缓存
@pytest.fixture           # 测试夹具
@dataclass                 # 数据类
@classmethod              # 类方法

很多人会用装饰器，却很难准确回答下面几个问题：

@decorator 到底在什么时候执行？
为什么装饰器里经常套三层函数？
为什么忘了 functools.wraps 会让 FastAPI、pytest 或调试器认错函数？
同步装饰器能不能直接套在 async def 上？
两个装饰器叠在一起时，谁先执行？
为什么一个看起来没问题的计数装饰器，到了多线程、多协程、多进程环境就不可信了？

先给出最重要的结论：

装饰器不是特殊类型，也不是编译器魔法。它只是一次对象替换：把原对象交给一个可调用对象，再把返回值绑定回原来的名字。

下面两种写法完全等价：

python

@measure_time
def query_user(user_id: int) -> dict:
    return {"id": user_id}

python

def query_user(user_id: int) -> dict:
    return {"id": user_id}

query_user = measure_time(query_user)

@ 只是把第二种写法放到了函数定义上方。理解了这行等价展开，装饰器最神秘的部分就已经消失了一半。

但项目里的问题通常不在“会不会写一个 wrapper”，而在于：这个替换发生在什么时候，替换后还保留了什么，额外状态由谁管理，它和其他装饰器又如何组合。

这篇文章就沿着这条线，从最小实现一路拆到项目级实践。

第一层：函数是一等对象，装饰器才有可能

Python 里的函数不只是“一段可以执行的代码”，它还是一个普通对象。它可以被赋值、放进容器、作为参数传入，也可以从另一个函数中返回：

python

from collections.abc import Callable


def greet(name: str) -> str:
    return f"你好，{name}"


# 赋值：变量指向同一个函数对象
say_hello = greet

# 放进容器
commands: dict[str, Callable[[str], str]] = {
    "hello": greet,
}

# 作为参数传递
def run(command: Callable[[str], str], value: str) -> str:
    return command(value)


print(say_hello("九老板"))
print(run(commands["hello"], "九老板"))

输出：

text

你好，九老板
你好，九老板

既然函数可以作为参数和返回值，那么我们就能写一个“接收函数、返回新函数”的函数：

python

def trace(func):
    def wrapper(*args, **kwargs):
        print(f"进入 {func.__name__}")
        result = func(*args, **kwargs)
        print(f"离开 {func.__name__}")
        return result

    return wrapper


@trace
def add(a: int, b: int) -> int:
    return a + b


print(add(2, 3))

输出：

text

进入 add
离开 add
5

这里发生了三件事：

Python 创建原始函数 add。
Python 执行 trace(add)，得到函数 wrapper。
名字 add 被重新绑定到 wrapper。

调用者以后执行的其实是 wrapper(2, 3)。wrapper 再在合适的时机调用原始 add，于是我们能在不修改原函数正文的情况下，在调用前后插入日志、计时、重试或权限检查。

闭包为什么能记住原函数

trace 已经执行完并返回了，为什么 wrapper 以后仍然能访问参数 func？因为 wrapper 形成了一个闭包（closure）。

闭包可以理解成“函数 + 它引用的外层环境”：

text

trace(add)
  │
  ├─ 局部变量 func ───────────────┐
  │                               │
  └─ 返回 wrapper                 │
         │                        │
         └─ 闭包继续引用 func ────┘

只要 wrapper 还活着，它引用的原始函数 func 就不会消失。装饰器能保存配置、原函数和少量状态，靠的正是闭包。

也可以直接观察闭包保存的内容：

python

original_add = add.__closure__[0].cell_contents
print(original_add)

不过这只是理解机制的实验手段。项目代码不要依赖 __closure__ 的位置去寻找原函数，后面会介绍标准的 __wrapped__ 和 inspect.unwrap()。

第二层：装饰发生在定义语句执行时，包装逻辑发生在调用时

这是装饰器在项目里最容易被忽略、也最重要的时间边界。

看下面这段代码：

python

def announce(label: str):
    print(f"创建装饰器：{label}")

    def decorator(func):
        print(f"装饰函数：{func.__name__}")

        def wrapper(*args, **kwargs):
            print(f"调用函数：{func.__name__}")
            return func(*args, **kwargs)

        return wrapper

    return decorator


@announce("订单服务")
def create_order() -> None:
    print("创建订单")


print("模块加载完成")
create_order()

输出顺序是：

text

创建装饰器：订单服务
装饰函数：create_order
模块加载完成
调用函数：create_order
创建订单

这里有两个完全不同的阶段。

阶段一：定义阶段

更准确地说，装饰器会在 Python 执行到被装饰的 def 或 class 语句时运行。模块顶层的定义通常发生在首次导入时；嵌套函数的定义，则可能在外层函数每次执行到那里时发生。

在当前例子中，模块第一次被导入、Python 执行到函数定义时：

text

announce("订单服务")
    ↓
decorator(create_order)
    ↓
create_order = wrapper

装饰器工厂和装饰动作都会在这个阶段发生。原函数的函数体不会执行，但装饰器外层的代码已经执行了。

阶段二：调用阶段

业务代码真正执行 create_order() 时，才会进入 wrapper，然后由 wrapper 决定是否、何时、以什么参数调用原函数。

阶段	发生时机	适合做什么	不适合做什么
装饰阶段	模块导入、类体执行、函数定义时	校验静态配置、构建轻量元数据、登记内存注册表	网络请求、连数据库、读取大文件、启动线程
调用阶段	每次调用被装饰函数时	日志、计时、权限判断、重试、事务边界	每次重复做可提前完成的重型反射

这带来几个直接的工程结论：

不要在装饰阶段执行外部 IO。 否则一次普通的 import 可能因为数据库或网络故障而失败。
静态校验应该尽早做。 @retry(attempts=0) 这种配置错误，最好在应用启动时就抛出，而不是等到第一笔流量到来。
注册型装饰器依赖模块被导入。 某个 handler 文件从未被 import，它里面的 @register 就从未执行。
装饰器参数通常会被冻结。 @retry(attempts=settings.RETRY_ATTEMPTS) 读取的是导入时的值，不会自动跟随运行时配置热更新。

装饰器可以有导入期行为，但应该让这种行为轻量、确定、无外部依赖。

第三层：`functools.wraps` 不是礼貌问题，而是函数契约

前面的 trace 能工作，但它悄悄破坏了原函数的身份：

python

import inspect

print(add.__name__)
print(add.__doc__)
print(inspect.signature(add))

可能得到：

text

wrapper
None
(*args, **kwargs)

原来的函数名、文档和签名都丢了。对人来说只是调试不方便，对框架来说可能直接变成功能错误：

FastAPI 依赖函数签名生成参数和 OpenAPI 文档。
pytest 根据函数标记和元数据发现测试或 fixture。
序列化、依赖注入、CLI 框架可能依赖 __module__、__qualname__ 或注解。
日志和 APM 最后只看到一堆名为 wrapper 的函数。

标准解法是 functools.wraps：

python

from functools import wraps


def trace(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"进入 {func.__qualname__}")
        try:
            return func(*args, **kwargs)
        finally:
            print(f"离开 {func.__qualname__}")

    return wrapper

@wraps(func) 会复制常用元信息，并在返回的 wrapper 上设置一个关键属性。重新用这个版本的 trace 装饰一个函数：

python

@trace
def add_with_wraps(a: int, b: int) -> int:
    return a + b


assert add_with_wraps.__wrapped__.__name__ == "add_with_wraps"

inspect.signature() 会沿着 __wrapped__ 找回原始签名，inspect.unwrap() 则能穿过多层装饰器拿到最里面的函数：

python

import inspect

original = inspect.unwrap(add_with_wraps)
assert original is add_with_wraps.__wrapped__
print(original.__name__)

输出是 add_with_wraps。这里用函数对象自身验证 __wrapped__，不要尝试在 trace 外部直接访问它的局部变量 wrapper。

只要装饰器返回了新的 callable，默认就应该使用 @wraps(func)。

少数不需要 wraps 的情况，是你明确在做“注册但不包装”：给原函数增加元数据后原样返回，或者故意把它转换成语义完全不同的对象。

`wraps` 保留运行时元数据，类型还要单独透传

下面这个注解看起来很常见：

python

def trace(func):
    ...

但类型检查器只知道它接收和返回“某个东西”，无法保证装饰前后的参数与返回值一致。项目里可以用 ParamSpec 和 TypeVar 描述这个契约：

python

import time
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def timed(func: Callable[P, R]) -> Callable[P, R]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        started_at = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            elapsed_ms = (time.perf_counter() - started_at) * 1000
            print(f"{func.__qualname__} took {elapsed_ms:.2f} ms")

    return wrapper

现在：

python

@timed
def load_user(user_id: int, *, include_orders: bool = False) -> dict:
    return {"id": user_id, "include_orders": include_orders}

类型检查器仍然知道 load_user 的参数是 user_id: int 和关键字参数 include_orders: bool，返回值是 dict。错误调用可以在开发阶段被发现：

python

load_user("not-an-int")  # 静态类型检查器可以报告错误

这里两套机制各管一层：

机制	解决什么问题
`@wraps(func)`	运行时的名称、文档、注解、`__wrapped__` 和反射签名
`ParamSpec` + `TypeVar`	静态类型系统中的参数列表与返回值透传

如果项目低于 Python 3.10，可以从 typing_extensions 导入 ParamSpec。

还要注意：wraps 让 inspect.signature() 能看到原签名，不代表 wrapper 在 CPython 调用边界真的拥有同样的形参。它实际仍接收 *args, **kwargs，参数校验最终由原函数完成。若装饰器有意增加、删除或改写参数，就不能假装签名没有变化，需要显式设计新的类型和反射签名。

第四层：带参数的装饰器为什么有三层函数

无参装饰器是这样：

python

@timed
def work():
    ...

等价于：

python

work = timed(work)

如果希望传配置：

python

@repeat(times=3)
def send_heartbeat() -> str:
    return "ok"

它的等价展开变成：

python

send_heartbeat = repeat(times=3)(send_heartbeat)

因此需要一个函数先接收配置并返回真正的装饰器：

python

from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def repeat(
    times: int,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    if times < 1:
        raise ValueError("times 必须大于等于 1")

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            result = func(*args, **kwargs)
            for _ in range(1, times):
                result = func(*args, **kwargs)
            return result

        return wrapper

    return decorator

三层函数分别承担三个时间点的职责：

text

repeat(times=3)                ← 导入时：读取并校验配置
    ↓ 返回 decorator

decorator(send_heartbeat)      ← 导入时：接收并保存原函数
    ↓ 返回 wrapper

wrapper()                       ← 调用时：执行包装逻辑和原函数

times 和 func 都被保存在 wrapper 的闭包中。装饰器工厂没有更神秘的东西，它只是把“接收配置”和“接收函数”拆成了两次调用。

什么时候应该让装饰器同时支持有参和无参？

有些库希望同时支持：

python

@tool
def search():
    ...


@tool(name="web_search")
def search_web():
    ...

这种 API 可以通过重载和判断第一个参数实现，但实现复杂度会明显上升，类型提示也更难写。项目内部如果没有强烈需求，最好选择一种明确形式：

python

@tool()
def search():
    ...


@tool(name="web_search")
def search_web():
    ...

为了省一对括号引入两套调用协议，往往不值得。

第五层：多个装饰器到底按什么顺序执行

装饰器可以叠加：

python

@outer
@inner
def work():
    ...

等价于：

python

work = outer(inner(work))

所以有三条顺序需要区分：

装饰器表达式从上往下求值：若写成 @outer() 和 @inner()，会先执行 outer()，再执行 inner()。
装饰时从下往上应用：先 inner(work)，再 outer(...)。
调用时从外往内进入、从内往外返回：先进入 outer 的 wrapper，再进入 inner 的 wrapper。

用一个例子观察完整顺序：

python

from functools import wraps


def layer(name: str):
    print(f"evaluate: {name}")

    def decorator(func):
        print(f"decorate: {name}")

        @wraps(func)
        def wrapper(*args, **kwargs):
            print(f"enter: {name}")
            try:
                return func(*args, **kwargs)
            finally:
                print(f"exit: {name}")

        return wrapper

    return decorator


@layer("outer")
@layer("inner")
def work():
    print("work")


work()

输出：

text

evaluate: outer
evaluate: inner
decorate: inner
decorate: outer
enter: outer
enter: inner
work
exit: inner
exit: outer

顺序不只是语法知识，它会改变业务语义。

重试与事务的顺序

python

@retry(attempts=3)
@transactional
def save_order():
    ...

等价于：

python

save_order = retry(attempts=3)(transactional(save_order))

每次重试都会重新进入 transactional，通常意味着每次尝试使用独立事务。

反过来：

python

@transactional
@retry(attempts=3)
def save_order():
    ...

所有重试可能都发生在同一个事务边界里。第一次失败已经把事务标记为不可用时，后面的尝试可能没有意义。

缓存与计时的顺序

python

@timed
@cache
def calculate():
    ...

所有调用都会经过 timed，缓存命中也会被记录。

python

@cache
@timed
def calculate():
    ...

缓存命中时直接从最外层返回，timed 根本不会执行，指标里只会出现缓存未命中的调用。

因此项目里不要把装饰器顺序当成排版问题。顺序就是调用链设计，至少应该在测试或文档中固定下来。

第六层：实例方法、类方法、静态方法和类装饰器

实例方法为什么通常可以直接装饰

python

class UserService:
    @timed
    def get_user(self, user_id: int) -> dict:
        return {"id": user_id}

这里的 wrapper 仍然是函数对象，也实现了描述符协议。通过实例访问时，Python 会自动把实例绑定为第一个参数：

python

service = UserService()
service.get_user(42)

# 概念上接近：
UserService.get_user(service, 42)

self 只是随着 *args 一起穿过 wrapper，最后传给原方法。

与 `classmethod`、`staticmethod`、`property` 的顺序

通用函数装饰器最好紧挨着 def，结构型装饰器放在外层：

python

class UserService:
    @classmethod
    @timed
    def from_config(cls, config: dict) -> "UserService":
        return cls()

    @staticmethod
    @timed
    def normalize_id(raw: str) -> int:
        return int(raw)

    @property
    @timed
    def service_name(self) -> str:
        return "user-service"

应用顺序是：

text

from_config = classmethod(timed(from_config))
normalize_id = staticmethod(timed(normalize_id))
service_name = property(timed(service_name))

这样 timed 接收到的是普通函数，外层再把它转换为类方法、静态方法或属性。反过来写时，通用装饰器收到的可能是 classmethod、staticmethod 或 property 描述符，而不是它预期的普通 callable。

类本身也可以被装饰

函数装饰器接收函数，类装饰器则接收类对象：

python

from collections.abc import Callable
from typing import TypeVar

T = TypeVar("T")
MODEL_REGISTRY: dict[str, type] = {}


def register_model(name: str) -> Callable[[type[T]], type[T]]:
    def decorator(cls: type[T]) -> type[T]:
        if name in MODEL_REGISTRY:
            raise ValueError(f"模型名重复：{name}")
        MODEL_REGISTRY[name] = cls
        return cls

    return decorator


@register_model("user")
class User:
    pass

它等价于：

python

class User:
    pass


User = register_model("user")(User)

标准库的 @dataclass 就是最典型的类装饰器。类装饰器适合做类级元数据、注册和有限的结构增强；如果需求涉及子类创建、继承链和属性解析的深度控制，再考虑 __init_subclass__ 或 metaclass，不要把所有元编程都塞进一个类装饰器。

装饰器本身也可以是对象

装饰器不一定由函数实现。只要对象可调用，也就是实现了 __call__，就能接收函数并返回包装对象：

python

from functools import update_wrapper


class CountCalls:
    def __init__(self, func):
        self.func = func
        self.calls = 0
        update_wrapper(self, func)

    def __call__(self, *args, **kwargs):
        self.calls += 1
        return self.func(*args, **kwargs)


@CountCalls
def ping() -> str:
    return "pong"


ping()
ping()
print(ping.calls)  # 2

类实现适合确实需要封装状态和多个辅助方法的装饰器，但要付出额外复杂度：calls 仍然不是线程安全或跨进程的全局计数；这个简单类也没有实现描述符 __get__，直接装饰实例方法时不会像普通函数 wrapper 那样自动绑定 self。因此无状态或轻状态场景优先使用闭包，只有对象模型能明显改善设计时再改用类。

第七层：同步、异步和生成器不能混为一谈

同步 wrapper 装饰 `async def`，可能只量到“创建协程”的时间

假设把前面的同步 @timed 直接套到异步函数上：

python

@timed
async def fetch_user(user_id: int) -> dict:
    ...

同步 wrapper 调用 func(*args, **kwargs) 时，得到的只是一个 coroutine 对象。真正的网络请求还没开始，它就已经结束计时并返回了。

更糟的是，外层 wrapper 是普通 def，一些依赖 inspect.iscoroutinefunction() 的框架可能把它误判为同步函数。@wraps 能保留名称和反射链，但不会把同步 wrapper 变成异步函数。

异步装饰器必须在异步 wrapper 里 await 原函数：

python

import time
from collections.abc import Awaitable, Callable
from functools import wraps
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def async_timed(
    func: Callable[P, Awaitable[R]],
) -> Callable[P, Awaitable[R]]:
    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        started_at = time.perf_counter()
        try:
            return await func(*args, **kwargs)
        finally:
            elapsed_ms = (time.perf_counter() - started_at) * 1000
            print(f"{func.__qualname__} took {elapsed_ms:.2f} ms")

    return wrapper

使用：

python

@async_timed
async def fetch_user(user_id: int) -> dict:
    return {"id": user_id}

这里的 finally 在成功、异常和任务取消时都会执行，因此计时能够覆盖完整生命周期。异步装饰器不要随便 except BaseException，否则可能误吞 KeyboardInterrupt、SystemExit 或任务取消信号。

如果一个公共库必须同时支持同步和异步函数，可以在装饰阶段用 inspect.iscoroutinefunction(func) 选择不同 wrapper。但项目内部通常更推荐分成 @timed 与 @async_timed：协议更清楚，类型更准确，也不容易让阻塞逻辑混入事件循环。

生成器调用返回的是迭代器，不代表数据已经消费

同样的问题也存在于生成器：

python

def stream_rows():
    yield from range(1_000_000)

普通 wrapper 执行 func() 时只创建生成器对象。如果想统计完整迭代过程，wrapper 自己也要参与迭代：

python

from collections.abc import Callable, Iterator
from functools import wraps
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
T = TypeVar("T")


def count_yields(
    func: Callable[P, Iterator[T]],
) -> Callable[P, Iterator[T]]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Iterator[T]:
        count = 0
        try:
            for item in func(*args, **kwargs):
                count += 1
                yield item
        finally:
            print(f"{func.__qualname__} yielded {count} items")

    return wrapper

异步生成器同理，需要 async for。还要注意：如果调用者没有消费完生成器，也没有显式关闭它，finally 的执行时间可能延后。上面的类型刻意写成 Iterator[T]，只承诺普通的拉取式迭代；如果原生成器的调用方还依赖 send() 或 throw()，这个 for 循环 wrapper 并不会把协议转发给内层生成器，需要专门实现完整的生成器协议。对流式响应、文件流和消息订阅做装饰时，必须围绕“消费生命周期”设计，而不是只围绕“创建对象”设计。

第八层：装饰器在真实项目里最适合做什么

装饰器最擅长处理的是横切关注点（cross-cutting concerns）：多个函数都需要、发生在稳定调用边界上、又不属于某个函数核心业务的逻辑。

场景一：注册——只增加元数据，不改变调用行为

事件系统、命令系统、插件系统和 Agent 工具系统经常使用注册型装饰器：

python

from collections.abc import Callable

Handler = Callable[[dict[str, object]], None]
HANDLERS: dict[str, Handler] = {}


def handles(event_name: str) -> Callable[[Handler], Handler]:
    def decorator(func: Handler) -> Handler:
        if event_name in HANDLERS:
            raise ValueError(f"事件处理器重复：{event_name}")
        HANDLERS[event_name] = func
        return func

    return decorator


@handles("order.created")
def send_order_notification(event: dict[str, object]) -> None:
    print(f"发送通知：{event}")

这个装饰器没有 wrapper，原函数身份和调用语义完全不变，只在导入时把引用放进注册表。

它的优点是声明和实现靠得很近；它的代价是依赖导入：

text

应用启动
  ↓
显式导入 handlers 包
  ↓
各模块执行 @handles(...)
  ↓
注册表完整

项目里要有一个清晰的 composition root 负责导入这些模块，不要期待 Python 自动扫描所有文件。注册动作也应该只改内存结构，不要在这里访问外部服务。

场景二：可观测性——记录结果，但不改变结果

日志、耗时和指标是典型的横切逻辑：

python

import logging
import time
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar

logger = logging.getLogger(__name__)
P = ParamSpec("P")
R = TypeVar("R")


def observed(
    operation: 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:
            started_at = time.perf_counter()

            try:
                result = func(*args, **kwargs)
            except Exception:
                elapsed_ms = (time.perf_counter() - started_at) * 1000
                logger.exception(
                    "operation_failed",
                    extra={
                        "operation": operation,
                        "elapsed_ms": elapsed_ms,
                    },
                )
                raise

            elapsed_ms = (time.perf_counter() - started_at) * 1000
            logger.info(
                "operation_succeeded",
                extra={
                    "operation": operation,
                    "elapsed_ms": elapsed_ms,
                },
            )
            return result

        return wrapper

    return decorator

这个实现有几个刻意的选择：

成功时原样返回结果，失败时用裸 raise 保留原异常和 traceback。
不把 args、kwargs 和返回值直接打进日志，避免泄露密码、Token、手机号等敏感信息。
用 time.perf_counter() 测量耗时，不用可能被系统校时影响的墙上时钟。
配置 operation 在导入时固定，单次调用的耗时在运行时计算。

真正项目里还可以把 trace ID 放在 contextvars.ContextVar 中，让装饰器读取当前请求上下文；不要用一个全局变量保存“当前请求”。异步函数则使用对应的 async wrapper。

场景三：重试——只重试明确、短暂、可恢复的失败

下面是一个同步重试装饰器的最小工程版本：

python

import logging
import time
from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar

logger = logging.getLogger(__name__)
P = ParamSpec("P")
R = TypeVar("R")


def retry(
    *,
    attempts: int = 3,
    retry_on: tuple[type[Exception], ...] = (TimeoutError,),
    base_delay: float = 0.2,
    max_delay: float = 2.0,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    if attempts < 1:
        raise ValueError("attempts 必须大于等于 1")
    if not retry_on:
        raise ValueError("retry_on 不能为空")
    if base_delay < 0 or max_delay < 0:
        raise ValueError("delay 不能小于 0")

    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 retry_on as exc:
                    if attempt == attempts:
                        raise

                    delay = min(
                        max_delay,
                        base_delay * (2 ** (attempt - 1)),
                    )
                    logger.warning(
                        "operation_retrying",
                        extra={
                            "function": func.__qualname__,
                            "attempt": attempt,
                            "next_delay": delay,
                            "error": repr(exc),
                        },
                    )
                    time.sleep(delay)

            raise AssertionError("unreachable")

        return wrapper

    return decorator

使用时只捕获明确的瞬时异常：

python

@retry(
    attempts=3,
    retry_on=(TimeoutError, ConnectionError),
    base_delay=0.1,
    max_delay=1.0,
)
def fetch_remote_config() -> dict:
    ...

生产重试还要回答四个问题：

操作幂等吗？ 查询通常可以重试，扣款、发消息、创建订单不能盲目重试，除非有幂等键。
哪些异常真的可恢复？ 参数错误和权限错误重试一百次也不会成功，不要写 retry_on=(Exception,)。
总超时预算是多少？ 三次单请求超时加两次退避，不能超过上层 HTTP 或任务 deadline。
会不会形成重试风暴？ 大规模服务通常还需要随机抖动、熔断、限流和服务端退避提示。

异步版本必须用 await asyncio.sleep(delay)，不能在 async def 的 wrapper 里调用 time.sleep()，否则会阻塞整个事件循环。成熟项目也可以直接采用 Tenacity 等经过验证的库，不必为复杂重试策略重新造轮子。

场景四：缓存——优先使用标准库

缓存也是装饰器的天然场景，但 Python 已经提供了成熟实现：

python

from functools import lru_cache


@lru_cache(maxsize=512)
def parse_rule(rule_text: str) -> tuple[str, ...]:
    return tuple(part.strip() for part in rule_text.split(","))

它适合参数可哈希、结果相对稳定、函数近似纯函数的场景。使用前仍要确认：

缓存如何失效？
参数和返回值会不会长期占用大量内存？
多进程 worker 之间是否允许各自保存一份缓存？
函数结果是否依赖时间、用户身份或外部状态？
异步函数返回的是 coroutine，能否直接缓存？通常不能照搬同步缓存。

装饰器让缓存看起来只有一行，但没有消除缓存一致性问题。

常见场景的适用度

场景	是否适合装饰器	关键边界
日志、计时、指标、trace	适合	不泄露参数，正确处理异常和 async
缓存	适合	纯度、失效、内存、多进程一致性
瞬时故障重试	有条件适合	幂等、异常白名单、退避、总超时
函数/插件/事件注册	适合	导入顺序、重复注册、只做内存操作
权限校验	有条件适合	上下文来源必须清楚，框架依赖机制可能更合适
数据库事务	有条件适合	嵌套事务、重试顺序、同步/异步 session
HTTP 全局日志与鉴权	通常不选	Web middleware 或依赖注入覆盖面更清晰
资源申请与释放	通常不选	`with` / `async with` 的生命周期更直观
业务流程和分支编排	不适合	应写成显式服务方法或工作流

判断标准不是“能不能写成装饰器”，而是“写成装饰器以后，调用者还能不能清楚地理解控制流和副作用”。

第九层：项目中最容易踩的坑

坑一：wrapper 吞掉异常或改变返回值

下面这种写法会把真实故障伪装成正常返回：

python

def unsafe(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception:
            return None

    return wrapper

调用方以后可能在很远的地方因为 None 再次报错，真正的异常栈已经丢失。除非装饰器的公开契约明确就是“失败返回默认值”，否则应该记录后重新抛出：

python

def wrapper(*args, **kwargs):
    try:
        return func(*args, **kwargs)
    except ExpectedError:
        logger.exception("operation_failed")
        raise

装饰器默认应满足：成功结果不变，失败类型不变，traceback 不丢。

坑二：闭包里的可变状态不是全局可靠状态

一个计数装饰器看起来很简单：

python

from functools import wraps


def count_calls(func):
    calls = 0

    @wraps(func)
    def wrapper(*args, **kwargs):
        nonlocal calls
        calls += 1
        print(f"第 {calls} 次调用")
        return func(*args, **kwargs)

    return wrapper

但 calls 的真实作用域是：当前进程中、当前被装饰函数对应的这个闭包。

多线程下，calls += 1 不是可靠的原子业务计数。
这段同步自增在单个事件循环里没有 await 切换点，但如果异步 wrapper 的“读取—修改—写回”跨过了 await，多协程同样会发生竞态。
多进程 worker 各有一份计数，任何一份都不是全局总数。
测试之间如果复用模块，状态可能相互污染。

调用次数、限流额度、熔断状态和业务统计应该放进线程安全对象、Redis、数据库或专业 metrics backend。闭包适合保存只读配置和原函数引用，不适合冒充共享状态系统。

坑三：装饰阶段做外部 IO

反例：

python

def load_remote_policy(func):
    policy = request_policy_service()  # import 时访问网络

    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, policy=policy, **kwargs)

    return wrapper

这会让模块导入依赖网络可用性，还把策略永久冻结在导入时。更好的方式是由应用启动流程显式加载依赖，再通过依赖注入、对象构造或调用期 provider 传入。

坑四：运行时配置被悄悄冻结

python

@retry(attempts=settings.RETRY_ATTEMPTS)
def call_service():
    ...

settings.RETRY_ATTEMPTS 在装饰阶段读取。即使后面修改 settings，闭包中的 attempts 也不会变化。

这不一定是错误。很多配置本来就应该在启动时固定，保持一次进程生命周期内行为一致。但如果需求明确要求动态配置，就应该把配置读取动作放到调用期，或注入一个显式 policy/provider，并在文档中说明成本与一致性语义。

坑五：装饰器隐藏了关键依赖

python

@require_permission("order:write")
@transactional
@publish_event("order.created")
def create_order(data):
    ...

看起来简洁，但函数实际上依赖用户上下文、数据库 session 和事件总线。这些对象从哪里来？事务失败后事件发没发？测试时怎么替换？如果答案只能靠读三个装饰器源码才能知道，抽象已经过头了。

关键业务依赖应该通过参数、构造函数或框架依赖注入显式表达。装饰器适合包住稳定边界，不适合把完整业务工作流藏起来。

坑六：装饰器层数太多，调用链变成“千层饼”

python

@feature_flag("new_checkout")
@authorize("order:create")
@rate_limit("checkout")
@retry(attempts=3)
@transactional
@audit("order.created")
@observed("create_order")
def create_order(data: dict):
    ...

每一层单独看都合理，叠在一起以后却很难回答：

鉴权失败会不会进入指标？
限流发生在重试外还是重试内？
审计记录每次尝试，还是只记录最终结果？
事务提交后发布事件，还是事件发布后才提交？

当装饰器开始表达业务顺序时，应该停下来改成显式 pipeline、service method 或 middleware chain。少量稳定横切层是抽象，多层业务控制流是隐藏。

坑七：只处理普通函数，忘了 async、生成器和描述符

一个装饰器准备上线前，至少要回答：

只支持普通 def，还是也支持 async def？
支持生成器和异步生成器吗？
能否装饰实例方法？
与 classmethod、staticmethod、property 的顺序是什么？
能否装饰 callable object？

不要用“应该能跑”代替明确协议。不支持的对象应在类型注解、文档或装饰阶段校验中直接说明。

坑八：以为 `wraps` 能解决一切

wraps 很重要，但它不会自动解决：

同步 wrapper 包异步函数；
新增或删除参数后的真实签名；
线程和协程安全；
多进程状态一致性；
缓存失效；
重试幂等性；
多层装饰器的业务顺序。

它解决的是元数据和反射链，不是完整语义。

第十层：在项目里怎样把装饰器用得更好

1. 先写出等价展开，再决定顺序

看到：

python

@A(x=1)
@B
@C(y=2)
def work():
    ...

先在脑中展开：

python

work = A(x=1)(B(C(y=2)(work)))

如果这个调用链不能被清楚解释，装饰器顺序就还没有设计完成。

2. 让 wrapper 尽量“透明”

一个基础设施装饰器最好保持以下契约：

text

参数：原样接收并转发
返回值：原样返回
异常：记录后原样抛出
元数据：使用 functools.wraps 保留
类型：使用 ParamSpec / TypeVar 透传
副作用：只增加文档中承诺的那一种

如果确实要改参数、返回值或异常，就把它当成一项正式 API 设计，而不是在 wrapper 里顺手转换。

3. 同步和异步协议分开设计

内部项目优先提供两个名字清楚的版本：

python

@observed("load_user")
def load_user(user_id: int):
    ...


@async_observed("load_user")
async def load_user_async(user_id: int):
    ...

这比一个内部充满 inspect、cast 和多分支的“万能装饰器”更容易维护。只有公共库确实需要统一 API 时，再增加同步/异步自动分派，并为两条路径分别测试。

4. 配置在装饰阶段校验，动态数据在调用阶段获取

静态配置：

python

@retry(attempts=3, base_delay=0.2)
def fetch():
    ...

可以在应用启动时校验，失败就尽早退出。

请求用户、trace ID、当前时间、实时开关等动态数据，则应在调用阶段从参数、ContextVar 或显式 provider 获取。不要在导入时捕获一个“当前用户”。

5. 把装饰器放在所属机制附近

不要急着创建一个无限膨胀的 utils/decorators.py。更清晰的组织方式是让装饰器靠近它服务的机制：

text

app/
├── observability/
│   ├── decorators.py      # observed、async_observed
│   └── metrics.py
├── resilience/
│   ├── retry.py           # retry policy 与装饰器
│   └── circuit_breaker.py
├── events/
│   ├── registry.py        # handles 与注册表
│   └── handlers/
└── orders/
    └── service.py

小项目里先把单次使用的装饰器放在当前模块。等它真的被多个模块复用、契约稳定后再提取，不要为了“看起来架构完整”提前建立抽象。

6. 优先复用标准库和框架能力

Python 自带了很多成熟装饰器：

装饰器	用途
`functools.cache` / `lru_cache`	记忆化缓存
`functools.cached_property`	实例属性惰性计算与缓存
`functools.singledispatch`	基于首参数类型的函数分派
`contextlib.contextmanager`	把生成器转换为上下文管理器
`dataclasses.dataclass`	生成数据类方法
`property`	把方法暴露为受控属性
`classmethod` / `staticmethod`	改变方法绑定方式

Web 请求级日志优先用 middleware，FastAPI 鉴权优先考虑 Depends，数据库生命周期优先使用 session/context manager。已有机制能更准确表达作用域时，不要为了使用装饰器而再包一层。

7. 为装饰器单独测试，也测试关键组合

装饰器的测试至少覆盖：

原函数成功时，参数和返回值不变。
原函数失败时，异常类型和 traceback 语义不变。
装饰器自己的副作用只发生预期次数。
__name__、__doc__、签名和 __wrapped__ 可用。
同步、异步或生成器协议分别正确。
与项目中允许叠加的其他装饰器顺序正确。

一个最小测试可以直接构造临时函数：

python

import inspect


def test_retry_retries_once() -> None:
    call_count = 0

    @retry(
        attempts=2,
        retry_on=(TimeoutError,),
        base_delay=0,
    )
    def flaky() -> str:
        nonlocal call_count
        call_count += 1
        if call_count == 1:
            raise TimeoutError("temporary")
        return "ok"

    assert flaky() == "ok"
    assert call_count == 2
    assert flaky.__name__ == "flaky"
    assert inspect.unwrap(flaky) is flaky.__wrapped__

调试多层装饰器时也可以使用：

python

original = inspect.unwrap(decorated_function)

但不要在正常业务里绕过 wrapper 调原函数，否则日志、权限、事务等契约会一起被绕过。

8. 关注热路径成本

每层 wrapper 都会增加一次 Python 函数调用。对 HTTP 请求、数据库访问和网络 IO 来说，这点成本通常可以忽略；对每秒调用数百万次的数值循环或序列化热路径，层层装饰可能变得可见。

常见优化原则：

可在装饰阶段完成的签名解析、正则编译和静态配置转换，不要每次调用重做。
日志级别关闭时，不要提前构造昂贵字符串或序列化大对象。
指标标签保持低基数，不要把 user ID、完整 URL 或异常文本当标签。
性能敏感时先 profile，再决定是否移除抽象，不要凭感觉优化。

装饰器、上下文管理器、中间件还是普通函数？

装饰器不是横切逻辑的唯一工具。可以用下面这张决策表快速判断：

你的需求	更合适的工具	原因
每次调用某个函数都要执行同一层逻辑	装饰器	边界稳定，调用方无需改变
只包围某一小段代码，而不是整个函数	`with` / `async with`	进入和退出位置显式可见
所有 HTTP 请求都要执行	middleware	作用域就是请求管线，不必逐个函数标记
参数解析、用户鉴权依赖请求上下文	框架依赖注入	依赖来源和生命周期更清楚
多步骤业务流程、有条件分支和补偿动作	显式 service/workflow	控制流不应藏在函数外层
只是某处调用前后各做一行操作	普通函数或直接代码	没有复用就不需要抽象
在导入时登记插件、命令、事件处理器	注册型装饰器	声明和实现靠近，语义稳定

一个简单判断方法是问自己：

去掉 @decorator 后，函数的业务含义是否仍然完整？

如果答案是“是，只是少了日志、缓存、重试或注册”，装饰器通常合适。

如果答案是“否，权限、事务、事件和核心流程都不见了”，说明装饰器正在隐藏业务，应该换成更显式的结构。

项目级最佳实践清单

最后把全文压缩成一份可以直接用于 Code Review 的清单：

先做等价展开：@decorator 就是 func = decorator(func)。
明确两个阶段：装饰发生在导入/定义时，wrapper 发生在调用时。
新 wrapper 默认使用 @wraps(func)，确保反射、文档和调试链完整。
用 ParamSpec 与 TypeVar 透传类型，不要让装饰后函数退化成 Any。
同步、异步、生成器分别设计，不要只验证“能返回一个对象”。
参数和返回值默认原样透传，异常默认原样抛出。
静态配置在装饰阶段校验，外部 IO 和请求数据留在调用阶段。
闭包优先保存只读配置，共享可变状态交给正确的并发与持久化设施。
把装饰器叠加顺序当成调用链设计，为关键组合写测试。
控制层数：横切逻辑可以包装，业务流程应该显式。
优先标准库、框架 middleware、依赖注入和 context manager，不要重复造轮子。
记录协议边界：支持哪些 callable、是否支持 async、有哪些副作用、如何与其他装饰器组合。

回顾：装饰器是“定义时组装，调用时代理”

把整个机制压缩成一张图：

text

定义 / 导入阶段

原始函数对象
    │
    ▼
inner(original)
    │
    ▼
outer(inner_wrapper)
    │
    ▼
函数名重新绑定到 outer_wrapper


调用阶段

调用者
  │
  ▼
outer wrapper：前置逻辑
  │
  ▼
inner wrapper：前置逻辑
  │
  ▼
原始函数
  │
  ▼
inner wrapper：后置逻辑
  │
  ▼
outer wrapper：后置逻辑
  │
  ▼
返回调用者

从语言层面看，装饰器只有一件事：

python

new_object = decorator(old_object)

从工程层面看，它同时涉及四份契约：

调用契约：参数、返回值和异常有没有变化？
反射契约：名称、文档、签名和类型有没有保留？
时间契约：哪些逻辑在导入时执行，哪些在调用时执行？
组合契约：多层装饰器按什么顺序进入、退出和处理失败？

真正用好装饰器，不是把更多代码藏到 @ 后面，而是只隐藏那些稳定、重复、与核心业务正交的调用边界。

最后记住一句话：装饰器不是魔法，它是在定义语句执行时完成的函数组装，也是调用时多出来的一层透明代理。透明，才是一个好装饰器最重要的品质。

从一个 @ 符号开始：它不是魔法

第一层：函数是一等对象，装饰器才有可能

闭包为什么能记住原函数

第二层：装饰发生在定义语句执行时，包装逻辑发生在调用时

阶段一：定义阶段

阶段二：调用阶段

第三层：functools.wraps 不是礼貌问题，而是函数契约

wraps 保留运行时元数据，类型还要单独透传

第四层：带参数的装饰器为什么有三层函数

什么时候应该让装饰器同时支持有参和无参？

第五层：多个装饰器到底按什么顺序执行

重试与事务的顺序

缓存与计时的顺序

第六层：实例方法、类方法、静态方法和类装饰器

实例方法为什么通常可以直接装饰

与 classmethod、staticmethod、property 的顺序

类本身也可以被装饰

装饰器本身也可以是对象

第七层：同步、异步和生成器不能混为一谈

同步 wrapper 装饰 async def，可能只量到“创建协程”的时间

生成器调用返回的是迭代器，不代表数据已经消费

第八层：装饰器在真实项目里最适合做什么

场景一：注册——只增加元数据，不改变调用行为

场景二：可观测性——记录结果，但不改变结果

场景三：重试——只重试明确、短暂、可恢复的失败

场景四：缓存——优先使用标准库

常见场景的适用度

第九层：项目中最容易踩的坑

坑一：wrapper 吞掉异常或改变返回值

坑二：闭包里的可变状态不是全局可靠状态

坑三：装饰阶段做外部 IO

坑四：运行时配置被悄悄冻结

坑五：装饰器隐藏了关键依赖

坑六：装饰器层数太多，调用链变成“千层饼”

坑七：只处理普通函数，忘了 async、生成器和描述符

坑八：以为 wraps 能解决一切

第十层：在项目里怎样把装饰器用得更好

1. 先写出等价展开，再决定顺序

2. 让 wrapper 尽量“透明”

3. 同步和异步协议分开设计

4. 配置在装饰阶段校验，动态数据在调用阶段获取

5. 把装饰器放在所属机制附近

6. 优先复用标准库和框架能力

7. 为装饰器单独测试，也测试关键组合

8. 关注热路径成本

装饰器、上下文管理器、中间件还是普通函数？

项目级最佳实践清单

回顾：装饰器是“定义时组装，调用时代理”

从一个 `@` 符号开始：它不是魔法

第三层：`functools.wraps` 不是礼貌问题，而是函数契约

`wraps` 保留运行时元数据，类型还要单独透传

与 `classmethod`、`staticmethod`、`property` 的顺序

同步 wrapper 装饰 `async def`，可能只量到“创建协程”的时间

坑八：以为 `wraps` 能解决一切