06-装饰器高级用法与最佳实践

本章是装饰器系列的终章，覆盖生产环境中最实用的高级技巧：

主题	核心概念	典型场景
精确类型签名	ParamSpec + Concatenate	类型安全的装饰器
async 装饰器	正确处理 await	FastAPI 端点装饰
类装饰器	`__call__` 协议	调用计数、单例模式
`__wrapped__` 调试	获取原始函数	测试、调试、自省
wrapt 库	第三方装饰器库	自动处理 self/async

概念铺垫

为什么需要高级装饰器用法？前面的章节我们掌握了基础装饰器的写法，但在生产环境中还需要解决：

类型安全：传统 Callable[..., T] 会丢失参数类型，IDE 无法正确提示
async 兼容：同步装饰器包裹异步函数会导致 coroutine 泄漏
状态管理：函数装饰器难以管理复杂状态，类装饰器更自然
调试追溯：多层装饰器难以定位原始函数和错误根源

L1 理解层：会用

精确类型签名：ParamSpec + Concatenate

为什么 `Callable[..., T]` 不够？

传统的装饰器类型签名使用 Callable[..., T]，这会导致类型信息丢失：

python

# ❌ 类型信息丢失
def bad_timer(func: Callable[..., T]) -> Callable[..., T]:
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> T:
        return func(*args, **kwargs)
    return wrapper

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

reveal_type(add)  # Callable[..., int] — 参数类型丢失！
add("x", "y")     # 类型检查器不会报错，但运行时会出错

ParamSpec 保留精确签名

ParamSpec（Python 3.10+）可以保留参数类型：

python

from typing import ParamSpec, TypeVar, Callable
import functools

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

def timer_precise(func: Callable[P, T]) -> Callable[P, T]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        return func(*args, **kwargs)
    return wrapper

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

reveal_type(add)  # (a: int, b: int) -> int — 完整签名保留！
add(1, 2)         # ✅ 类型检查通过
add("x", "y")     # ❌ 类型检查报错

Concatenate 添加/移除参数

Concatenate（Python 3.10+）用于在装饰器中添加或删除参数：

python

from typing import Concatenate

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

# 添加一个 ctx 参数
def with_context(
    func: Callable[Concatenate[str, P], R]
) -> Callable[P, R]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        ctx = "default-context"
        return func(ctx, *args, **kwargs)
    return wrapper

@with_context
def process(ctx: str, data: int) -> str:
    return f"{ctx}: {data}"

process(42)  # 调用时不需要传 ctx

类型签名对比

方式	参数类型保留	IDE 补全	类型检查
`Callable[..., T]`	❌ 丢失	❌ 无	❌ 不严格
`Callable[P, T]`	✅ 保留	✅ 完整	✅ 严格
`Concatenate[X, P]`	✅ 修改	✅ 完整	✅ 严格

async 装饰器：正确处理 await

常见错误：不用 await 的 bug

python

# ❌ 错误：忘记 await，返回的是 coroutine 对象而不是结果
def broken_async_timer(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)  # 没有 await！
        elapsed = time.perf_counter() - start
        return result  # 返回的是 coroutine，不是实际结果
    return wrapper

调用者会得到一个 coroutine 对象而不是预期的值：

python

@broken_async_timer
async def fetch_data():
    return {"status": "ok"}

result = await fetch_data()
# result 是 coroutine 对象，不是 dict！

正确写法

python

# ✅ 正确：使用 await
from collections.abc import Coroutine
from typing import Any, ParamSpec, TypeVar, Callable

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

def async_timer(
    func: Callable[P, Coroutine[Any, Any, T]],
) -> Callable[P, Coroutine[Any, Any, T]]:
    @functools.wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        start = time.perf_counter()
        result = await func(*args, **kwargs)  # 正确 await
        elapsed = time.perf_counter() - start
        wrapper._last_elapsed = elapsed
        return result
    wrapper._last_elapsed = 0.0
    return wrapper

FastAPI 中的 async 装饰器

FastAPI 路由经常使用 async 装饰器。本章的 app/routers/async_ops.py 提供了演示：

bash

# 异步端点 — 装饰器正确 await
$ curl "http://localhost:8000/api/v1/async/call/100"
{
  "delay_ms": 100,
  "status": "done",
  "elapsed_ms": "100.23"
}

# 同步端点 — 对比
$ curl "http://localhost:8000/api/v1/async/sync/100"
{
  "delay_ms": 100,
  "status": "done",
  "elapsed_ms": "100.45"
}

类装饰器：`call` 协议

类装饰器 vs 函数装饰器

类装饰器通过实现 __call__ 方法工作：

python

class CountCalls:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self._func = func
        self._count = 0

    def __call__(self, *args, **kwargs):
        self._count += 1
        return self._func(*args, **kwargs)

    @property
    def count(self):
        return self._count

对比表

特性	函数装饰器	类装饰器
语法	`def decorator(func):`	`class Decorator:`
状态存储	闭包变量 / 函数属性	实例属性
可读性	简洁	更 OOP
类型提示	需要 ParamSpec	更灵活
带参数	需要三层嵌套	`__init__` 接收参数
方法	需要手动处理	天然支持

单例模式装饰器

python

class Singleton:
    def __init__(self, cls: type) -> None:
        self._cls = cls
        self._instance: Any = None
        functools.update_wrapper(self, cls)

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        if self._instance is None:
            self._instance = self._cls(*args, **kwargs)
        return self._instance

    def reset(self) -> None:
        self._instance = None

# 使用
@Singleton
class Config:
    def __init__(self, dsn: str) -> None:
        self.dsn = dsn

c1 = Config("redis://localhost")
c2 = Config("redis://other")  # 返回同一个实例
assert c1 is c2
assert c1.dsn == "redis://localhost"  # 第一次初始化的值

`wrapped` 的调试技巧

如何获取被装饰的原始函数

functools.wraps 会设置 __wrapped__ 属性指向原始函数：

python

def get_original_func(decorated_func: Callable) -> Callable:
    return getattr(decorated_func, "__wrapped__", decorated_func)

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

original = get_original_func(add)
assert original is not add  # 是不同的对象
assert original(1, 2) == 3  # 但功能一样

如何检测函数是否被装饰过

python

def is_decorated(func: Callable) -> bool:
    return hasattr(func, "__wrapped__")

@timer_precise
def decorated():
    pass

def plain():
    pass

assert is_decorated(decorated) is True
assert is_decorated(plain) is False

调试场景

python

import inspect

def debug_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 获取原始函数的源码位置
        original = getattr(func, "__wrapped__", func)
        source_file = inspect.getfile(original)
        source_lines, start_line = inspect.getsourcelines(original)
        print(f"Function {original.__name__} defined at {source_file}:{start_line}")
        return func(*args, **kwargs)
    return wrapper

wrapt 库简介

wrapt 是第三方装饰器库，自动处理 self、async 等痛点：

安装

bash

uv add wrapt

为什么需要 wrapt？

痛点	原生方案	wrapt 方案
装饰方法丢失 self	手动处理	自动处理
async 装饰器	单独写 async 版本	自动检测
带参数装饰器	三层嵌套	统一接口
类型提示	复杂	更简洁

使用示例

python

import wrapt

@wrapt.decorator
def universal_timer(wrapped, instance, args, kwargs):
    # wrapped: 被装饰的函数
    # instance: 如果是方法装饰，则是 self；否则是 None
    # args, kwargs: 调用参数
    start = time.perf_counter()
    result = wrapped(*args, **kwargs)
    elapsed = time.perf_counter() - start
    print(f"{wrapped.__name__} took {elapsed:.4f}s")
    return result

# 同时适用于函数和方法
@universal_timer
def sync_func(x):
    time.sleep(0.1)

class MyClass:
    @universal_timer
    def method(self, x):
        time.sleep(0.1)

贯穿实战：异步端点 + 调用计数

本章的实战代码位于 app/routers/async_ops.py，演示了：

异步计时端点

GET /api/v1/async/call/{delay_ms} — 使用 @async_timer 装饰的异步操作，正确 await 内部 coroutine。

同步计时端点

GET /api/v1/async/sync/{delay_ms} — 使用 @timer_precise 装饰的同步操作，用于性能对比。

调用计数端点

GET /api/v1/async/count-calls — 演示 @CountCalls 类装饰器，每次请求都会创建新的计数函数并调用 4 次，返回 total_calls: 4。

L2 实践层：用好

做法	原因	示例
Python 3.10+ 使用 ParamSpec	保留完整函数签名，IDE 类型检查	`func: Callable[P, T] -> Callable[P, T]`
始终使用 `@functools.wraps`	保留 `__name__`、`__doc__`、`__module__`	`@functools.wraps(func)`
async 装饰器必须 await	否则返回 coroutine 对象而非结果	`result = await func(args, *kwargs)`
类装饰器使用 `functools.update_wrapper`	保留元数据，等效于函数装饰器的 @wraps	`functools.update_wrapper(self, func)`
装饰器应该是幂等的	多次装饰不产生副作用	见下方反模式中的示例
复杂装饰器考虑使用 wrapt	自动处理 self/async，减少代码量	`@wrapt.decorator`
使用 `Concatenate` 修改签名	装饰器添加/移除参数时类型安全	`Callable[Concatenate[str, P], R]`

反模式：不要这样做

python

# ❌ 错误：装饰器不是幂等的 — 每次装饰都重复注册
REGISTRY = []
def register(func):
    REGISTRY.append(func)  # 次次追加，重复装饰会多次添加
    return func
# ✅ 正确：使用字典或检查去重
REGISTRY = {}
def register(func=None, *, name=None):
    def decorator(f):
        key = name or f.__name__
        REGISTRY[key] = f  # 字典去重
        return f
    return decorator(func) if func else decorator

# ❌ 错误：async 装饰器忘记 await
def async_decorator(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)  # 没有 await！
        return result
    return wrapper

# ❌ 错误：类装饰器不用 update_wrapper
class BadCounter:
    def __init__(self, func):
        self._func = func  # 忘记 update_wrapper
        self._count = 0
# ✅ 正确：
class GoodCounter:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self._func = func
        self._count = 0

# ❌ 错误：在装饰器层面做耗时 I/O（定义阶段执行）
def bad_cache_decorator(func):
    data = requests.get("https://api.example.com/config")  # 定义时网络请求！
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs, config=data)
    return wrapper

适用场景

场景	是否推荐	原因
需要类型安全的装饰器	✅ 推荐 ParamSpec	保留完整签名，IDE 可用
修改函数签名（添加/移除参数）	✅ 推荐 Concatenate	类型检查可验证
async FastAPI 端点	✅ 推荐 async 装饰器	正确处理 await，不泄漏 coroutine
需要维护状态的装饰器	✅ 推荐类装饰器	实例属性比闭包更清晰
单例模式	✅ 推荐类装饰器	`__init__` + `__call__` 天然适合
同时装饰函数和方法	✅ 推荐 wrapt	自动处理 self，统一接口
简单无状态装饰器	⚠️ 函数装饰器即可	类装饰器过于重型
Python 3.9 及以下	❌ 无法用 ParamSpec	3.10+ 才支持

装饰器最佳实践清单

1. 始终使用 `@functools.wraps`

保留原始函数的 __name__、__doc__、__module__ 等元数据：

python

import functools

def my_decorator(func):
    @functools.wraps(func)  # ← 必须
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

2. 使用 ParamSpec 保留类型签名

Python 3.10+ 项目中使用 ParamSpec 而不是 Callable[..., T]：

python

from typing import ParamSpec, TypeVar, Callable

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

def decorator(func: Callable[P, T]) -> Callable[P, T]:
    ...

3. async 装饰器必须 await

python

def async_decorator(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)  # ← 必须 await
    return wrapper

4. 类装饰器使用 `functools.update_wrapper`

python

class ClassDecorator:
    def __init__(self, func):
        functools.update_wrapper(self, func)  # ← 保留元数据
        self._func = func

5. 装饰器应该是幂等的

多次装饰同一函数不应产生副作用：

python

# ❌ 不好：每次装饰都注册到全局列表
def register(func):
    REGISTRY.append(func)
    return func

# ✅ 好：使用装饰器参数控制
def register(func=None, *, name=None):
    def decorator(f):
        REGISTRY[name or f.__name__] = f
        return f
    return decorator(func) if func else decorator

6. 复杂装饰器考虑使用 wrapt

当装饰器需要同时支持函数、方法、async 函数时，使用 wrapt 可以大幅减少代码量：

python

import wrapt

@wrapt.decorator
def my_decorator(wrapped, instance, args, kwargs):
    return wrapped(*args, **kwargs)

自检清单

完成本章后，你应该能够：

[ ] 使用 ParamSpec 编写类型安全的装饰器
[ ] 使用 Concatenate 修改函数签名
[ ] 编写正确的 async 装饰器（正确处理 await）
[ ] 使用类装饰器实现 __call__ 协议
[ ] 实现单例模式装饰器
[ ] 通过 __wrapped__ 获取原始函数
[ ] 检测函数是否被装饰过
[ ] 理解 wrapt 库的价值和使用场景
[ ] 遵循装饰器最佳实践清单

能力清单

[ ] 理解 Callable[..., T] 与 Callable[P, T] 的区别
[ ] 能够在 FastAPI 项目中安全使用 async 装饰器
[ ] 能够选择函数装饰器或类装饰器的合适场景
[ ] 能够调试被装饰的函数（__wrapped__、inspect 模块）
[ ] 能够评估何时引入 wrapt 库

06-装饰器高级用法与最佳实践 ​

概念铺垫 ​

L1 理解层：会用 ​

精确类型签名：ParamSpec + Concatenate ​

为什么 Callable[..., T] 不够？ ​

ParamSpec 保留精确签名 ​

Concatenate 添加/移除参数 ​

类型签名对比 ​

async 装饰器：正确处理 await ​

常见错误：不用 await 的 bug ​

正确写法 ​

FastAPI 中的 async 装饰器 ​

类装饰器：__call__ 协议 ​

类装饰器 vs 函数装饰器 ​

对比表 ​

单例模式装饰器 ​

__wrapped__ 的调试技巧 ​

如何获取被装饰的原始函数 ​

如何检测函数是否被装饰过 ​

调试场景 ​

wrapt 库简介 ​

安装 ​

为什么需要 wrapt？ ​

使用示例 ​

贯穿实战：异步端点 + 调用计数 ​

异步计时端点 ​

同步计时端点 ​

调用计数端点 ​

L2 实践层：用好 ​

推荐做法 ​

反模式：不要这样做 ​

适用场景 ​

装饰器最佳实践清单 ​

1. 始终使用 @functools.wraps ​

2. 使用 ParamSpec 保留类型签名 ​

3. async 装饰器必须 await ​

4. 类装饰器使用 functools.update_wrapper ​

5. 装饰器应该是幂等的 ​

6. 复杂装饰器考虑使用 wrapt ​

自检清单 ​

能力清单 ​

延伸阅读 ​

06-装饰器高级用法与最佳实践

概念铺垫

L1 理解层：会用

精确类型签名：ParamSpec + Concatenate

为什么 `Callable[..., T]` 不够？

ParamSpec 保留精确签名

Concatenate 添加/移除参数

类型签名对比

async 装饰器：正确处理 await

常见错误：不用 await 的 bug

正确写法

FastAPI 中的 async 装饰器

类装饰器：`call` 协议

类装饰器 vs 函数装饰器

对比表

单例模式装饰器

`wrapped` 的调试技巧

如何获取被装饰的原始函数

如何检测函数是否被装饰过

调试场景

wrapt 库简介

安装

为什么需要 wrapt？

使用示例

贯穿实战：异步端点 + 调用计数

异步计时端点

同步计时端点

调用计数端点

L2 实践层：用好

推荐做法

反模式：不要这样做

适用场景

装饰器最佳实践清单

1. 始终使用 `@functools.wraps`

2. 使用 ParamSpec 保留类型签名

3. async 装饰器必须 await

4. 类装饰器使用 `functools.update_wrapper`

5. 装饰器应该是幂等的

6. 复杂装饰器考虑使用 wrapt

自检清单

能力清单

延伸阅读