04-带参数装饰器与装饰器工厂
上一章我们用两层嵌套写好了基础装饰器，但有个问题：装饰器无法接收自定义参数。如果想让装饰器"可配置"，该怎么办？
概念铺垫
1. 为什么需要带参数装饰器？
基础装饰器的局限：
python
# ❌ 基础装饰器 — 无法配置
@log_calls
def foo(): ...
# 想要这样写，但语法不允许
@log_calls(level="debug", output="file")
def foo(): ...问题：装饰器本身接收的参数是"被装饰的函数"，没有多余的位置接收配置参数。
解决方案：再加一层嵌套——用外层函数接收配置，内层函数接收被装饰的函数。
2. 生活类比：定制包装盒
| 层次 | 类比 | 代码层 |
|---|---|---|
| 第 1 层 | 告诉工厂"我要多大的盒子"（尺寸参数） | 参数层 — 接收 times、max_attempts |
| 第 2 层 | 工厂根据你的要求制作包装盒 | 装饰器层 — 接收 func |
| 第 3 层 | 包装盒把商品包裹起来 | 包装层 — wrapper 包裹函数调用 |
参数层(配置) → 装饰器层(函数) → 包装层(调用)L1 理解层：会用
3. 三层嵌套结构 — 逐层拆解
以 repeat 装饰器为例：
python
def repeat(times: int) -> Callable[[Callable[..., T]], Callable[..., T]]:
# ── 第 1 层：参数层 ──
# 接收配置参数 times
def decorator(func: Callable[..., T]) -> Callable[..., T]:
# ── 第 2 层：装饰器层 ──
# 接收被装饰的函数 func
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> T:
# ── 第 3 层：包装层 ──
# 实际执行逻辑
result = None
for _ in range(times): # ← 这里用了第 1 层的参数
result = func(*args, **kwargs) # ← 这里调用了第 2 层的函数
return result
return wrapper
return decorator调用过程：
python
@repeat(times=3) # 第 1 层被调用，返回 decorator
def greet(): # 第 2 层被调用，decorator(greet) 返回 wrapper
print("hello") # 每次调用 greet 实际执行的是 wrapper
greet() # wrapper 执行 3 次关键理解
python
@repeat(times=3)
def greet(): ...
# 等价于：
greet = repeat(times=3)(greet)
# ──────┬─────┬──────
# │ └── 第 2 层：decorator(greet) → wrapper
# └── 第 1 层：repeat(3) → decorator4. 类型签名：Callable[[Callable[..., T]], Callable[..., T]]
这个类型出现在第 2 层（装饰器层）的返回值上：
python
def decorator(func: Callable[..., T]) -> Callable[..., T]:拆解：
| 部分 | 含义 |
|---|---|
Callable[..., T] | 输入：任意参数、返回 T 的函数 |
Callable[..., T] | 输出：任意参数、返回 T 的函数 |
| 整体 | "接收一个函数，返回一个相同签名的函数" |
加上第 1 层后，完整签名是：
python
def repeat(times: int) -> Callable[[Callable[..., T]], Callable[..., T]]:times: int→ 第 1 层的参数- 返回值 → 一个装饰器（第 2 层），类型是
Callable[[Callable[..., T]], Callable[..., T]]
5. 贯穿实战 1：权限验证装饰器
装饰器定义
python
def require_role(required_role: str) -> Callable:
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
req = kwargs.get("request") or (args[0] if args else None)
user_role = getattr(req, "role", "guest") if req else "guest"
if user_role != required_role:
return {"status": 403, "error": "权限不足"}
return func(*args, **kwargs)
return wrapper
return decoratorFastAPI 路由集成
python
# app/routers/auth.py
@require_role("admin")
def _delete_user(user_id: int, request: RequestContext) -> dict:
return {"status": 200, "deleted": user_id}
@router.delete("/users/{user_id}")
def api_delete_user(user_id: int, x_role: str = Header(default="guest")) -> dict:
return _delete_user(user_id, request=RequestContext(role=x_role))curl 示例
bash
# ✅ 管理员删除用户
curl -X DELETE http://localhost:8000/api/v1/auth/users/1 \
-H "X-Role: admin"
# {"status": 200, "deleted": 1}
# ❌ 访客无权删除
curl -X DELETE http://localhost:8000/api/v1/auth/users/1 \
-H "X-Role: guest"
# {"status": 403, "error": "权限不足", "required": "admin", "actual": "guest"}6. 贯穿实战 2：重试装饰器
装饰器定义
python
def retry(
max_attempts: int = 3,
delay: float = 0.0,
exceptions: tuple[type[Exception], ...] = (Exception,),
) -> Callable[[Callable[..., T]], Callable[..., T]]:
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> T:
last_exc = None
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exc = e
if delay > 0 and attempt < max_attempts:
time.sleep(delay)
raise last_exc
return wrapper
return decoratorFastAPI 路由集成
python
# app/routers/retry.py
@retry(max_attempts=3, delay=0.0, exceptions=(ConnectionError,))
def _unstable_api() -> dict:
global _attempt_count
_attempt_count += 1
if _attempt_count < 3:
raise ConnectionError("连接失败")
return {"status": "success", "attempts": _attempt_count}bash
curl -X POST http://localhost:8000/api/v1/retry/fetch
# {"status": "success", "attempts": 3}7. 装饰器工厂模式
装饰器工厂 = 一个返回装饰器的函数，可以根据配置生成不同的装饰器实例：
python
def create_rate_limiter(max_calls: int) -> Callable:
def decorator(func: Callable) -> Callable:
calls = 0
@functools.wraps(func)
def wrapper(*args, **kwargs):
nonlocal calls
if calls >= max_calls:
raise RuntimeError(f"调用次数超限（上限 {max_calls}）")
calls += 1
return func(*args, **kwargs)
def reset():
nonlocal calls
calls = 0
wrapper.reset = reset
wrapper.call_count = lambda: calls
return wrapper
return decorator使用示例
python
# 创建不同限制的装饰器
strict_limiter = create_rate_limiter(3)
relaxed_limiter = create_rate_limiter(100)
@strict_limiter
def api_call():
return "ok"
api_call() # OK (1/3)
api_call() # OK (2/3)
api_call() # OK (3/3)
api_call() # RuntimeError: 调用次数超限（上限 3）
# 重置计数器
api_call.reset()
api_call() # OK (1/3)工厂模式的优势
| 优势 | 说明 |
|---|---|
| 配置隔离 | 每个装饰器实例有独立的 calls 状态 |
| 可复用 | 同一工厂可创建不同参数的装饰器 |
| 可扩展 | 可以附加 reset、call_count 等方法 |
8. 三层嵌套 vs 两层嵌套对比
| 特性 | 两层嵌套（第 3 章） | 三层嵌套（第 4 章） |
|---|---|---|
| 结构 | decorator → wrapper | 参数层 → decorator → wrapper |
| 参数 | 无自定义参数 | 可接收配置参数 |
| 用法 | @log_calls | @retry(max_attempts=3) |
| 类型 | Callable[[T], T] | 返回 Callable[[T], T] 的函数 |
| 适用场景 | 通用日志、计时 | 权限、重试、限流 |
L2 实践层：用好
推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 三层嵌套实现带参数装饰器 | 标准模式，清晰分离配置/函数/调用 | def retry(max): def decorator(func): def wrapper: |
| 参数设置默认值 | 支持无参数调用 @decorator 和 @decorator() | max_attempts: int = 3 |
| 用 dataclass 传复杂参数 | 参数多时提高可读性 | @auth_required(AuthConfig(role="admin", scope="all")) |
| 附加方法到 wrapper | 提供 reset/call_count 等控制接口 | wrapper.reset = reset |
| 捕获特定异常 | 避免掩盖无关错误 | exceptions=(ConnectionError,) |
| 支持无括号调用 | 让装饰器同时支持 @deco 和 @deco() | 见下方反模式中的修正方案 |
反模式：不要这样做
python
# ❌ 错误：用两层嵌套伪装带参数（hack 方式检测参数）
def bad_repeat(func=None, *, times=1):
def decorator(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
result = None
for _ in range(times):
result = f(*args, **kwargs)
return result
return wrapper
if func is None:
return decorator
return decorator(func)
# ✅ 正确：标准三层嵌套
def repeat(times: int = 1):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
result = None
for _ in range(times):
result = func(*args, **kwargs)
return result
return wrapper
return decorator
# ❌ 错误：装饰器工厂闭包引用可变配置（修改后影响已装饰的函数）
config = {"max": 5}
def create_limiter():
max_calls = config["max"] # 定义时捕获当前值 5
def decorator(func):
calls = 0
def wrapper(*args, **kwargs):
nonlocal calls
if calls >= max_calls:
raise RuntimeError
calls += 1
return func(*args, **kwargs)
return wrapper
return decorator
# 后续 config["max"] = 3 不会影响已装饰的 limiter！
# ✅ 正确：如果需要动态响应，在 wrapper 内读取
# ❌ 错误：装饰器参数是可变的默认可变参数
def register(func, registry=[]): # 默认可变参数共享！
registry.append(func.__name__)
return func
# ✅ 正确：
def register(func, registry=None):
if registry is None:
registry = []
registry.append(func.__name__)
return func适用场景
| 场景 | 是否推荐 | 原因 |
|---|---|---|
| 权限验证（RBAC） | ✅ 推荐 | @require_role("admin") 清晰表达角色要求 |
| API 重试 | ✅ 推荐 | @retry(max_attempts=3, delay=0.5) 声明式重试 |
| 速率限制 | ✅ 推荐 | @rate_limit(max_calls=100) 声明式限流 |
| 参数验证 | ✅ 推荐 | @validate(schema=UserSchema) 声明式校验 |
| 简单开关 | ✅ 推荐 | @feature_flag("new_checkout") 声明式特性开关 |
| 状态机 | ⚠️ 可选 | 闭包状态 + 装饰器可以实现，但类可能更清晰 |
| 需要持久化配置 | ❌ 不推荐 | 装饰器配置在模块加载时确定，运行时不改变 |
L3 专家层：深入
Python 如何实现：三层嵌套闭包链与参数存储
带参数装饰器的本质是三层闭包嵌套，每一层捕获不同的上下文：
┌─────────────────────────────────────────────────────────────┐
│ 三层嵌套的闭包链 │
│ │
│ def retry(max_attempts=3, delay=0.0): │
│ def decorator(func): │
│ @functools.wraps(func) │
│ def wrapper(*args, **kwargs): │
│ # 这里可以访问 max_attempts, delay, func │
│ ... │
│ return wrapper │
│ return decorator │
│ │
│ @retry(max_attempts=5, delay=1.0) │
│ def fetch_data(): │
│ ... │
│ │
│ 内存中的对象关系： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ 第 1 层：retry(max_attempts=5, delay=1.0) │ │
│ │ 返回 decorator 函数 │ │
│ │ • decorator.__closure__ │ │
│ │ ┌──────────────┬──────────┐ │ │
│ │ │ cell 0 │ cell 1 │ │ │
│ │ │ max_attempts │ delay │ │ │
│ │ │ = 5 │ = 1.0 │ │ │
│ │ └──────────────┴──────────┘ │ │
│ │ │ │
│ │ 第 2 层：decorator(fetch_data) │ │
│ │ 返回 wrapper 函数 │ │
│ │ • wrapper.__closure__ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ cell 0 │ │ │
│ │ │ func (fetch_data) │ │ │
│ │ └──────────────────────┘ │ │
│ │ │ │
│ │ 第 3 层：wrapper(*args, **kwargs) │ │
│ │ • 无 __closure__（如果内部不创建冒泡引用） │ │
│ │ • 从第 2 层 cell 读取 func │ │
│ │ • 从第 1 层 cell 读取 max_attempts, delay │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ wrapper 通过 LEGB 查找访问变量： │
│ • func → E 层（decorator 的作用域，通过 cell 间接引用） │
│ • max_attempts → 通过 decorator 的闭包间接访问 │
│ • delay → 通过 decorator 的闭包间接访问 │
│ │
└─────────────────────────────────────────────────────────────┘验证：检查装饰器工厂的闭包链
python
def retry(max_attempts=3, delay=0.0):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
pass
return wrapper
return decorator
@retry(max_attempts=5, delay=1.0)
def fetch_data():
pass
# 第 1 层闭包：decorator 捕获的变量
decorator = retry(max_attempts=5, delay=1.0) # 等价于 @ 过程中间步骤
print(decorator.__closure__) # (cell, cell)
print(decorator.__closure__[0].cell_contents) # 5 (max_attempts)
print(decorator.__closure__[1].cell_contents) # 1.0 (delay)
# 第 2 层闭包：wrapper 捕获的变量
print(fetch_data.__closure__) # (cell,)
print(fetch_data.__closure__[0].cell_contents) # <function fetch_data> (func)
# wrapper 本身是顶层函数（无闭包，但通过闭包链间接访问）
print(fetch_data.__code__.co_freevars) # ('func',)偏函数应用模式（Partial Application Pattern）
带参数装饰器本质上是偏函数应用的一种形式：
┌─────────────────────────────────────────────────────────────┐
│ 装饰器工厂 = 偏函数应用模式 │
│ │
│ retry(max_attempts=5) = partial(retry_impl, max_attempts=5)│
│ │
│ 分步应用： │
│ step1 = retry(max_attempts=5, delay=1.0) ← 固定配置参数 │
│ step2 = step1(fetch_data) ← 应用被装饰函数 │
│ 等价于：retry(max_attempts=5, delay=1.0)(fetch_data) │
│ │
│ 这等同于柯里化（Currying）： │
│ retry :: Config -> Decorator :: (a -> b) -> (a -> b) │
│ 其中 Config = (max_attempts: int, delay: float) │
│ │
└─────────────────────────────────────────────────────────────┘性能考量
| 操作 | 时间复杂度 | 说明 |
|---|---|---|
@retry(max_attempts=3) 定义阶段 | O(1) | 第 1 层调用 1 次 + 第 2 层调用 1 次 |
wrapper() 每次调用 | O(1) | 与两层装修器相近，LEGB 查找多跳一层 |
nonlocal 修改闭包状态 | O(1) | 通过 cell 对象间接写入 |
| 工厂模式创建多个实例 | O(n) | n = 被装饰函数数量 |
| 场景 | 两层装饰器 | 三层装饰器 | 说明 |
|---|---|---|---|
| 定义阶段开销 | ~200ns | ~300ns | 三层多一次函数调用 |
| 调用阶段开销 | ~90ns | ~100ns | LEGB 查找多跳一层（~10ns） |
| 内存占用 | ~72 bytes | ~120 bytes | 额外一层闭包 + cell 元组 |
知识关联
┌─────────────────────────────────────────────────────────────┐
│ 知识关联图：带参数装饰器 → 高级模式 │
│ │
│ 第 4 章：带参数装饰器与装饰器工厂 │
│ ┌──────────────────────────────────────┐ │
│ │ • 三层嵌套 = 参数层 + 装饰器层 + 包装层│ │
│ │ • 装饰器工厂 = 返回装饰器的函数 │ │
│ │ • 闭包链存储各级参数 │ │
│ └──────────┬───────────────────────────┘ │
│ │ │
│ ┌────────┼──────────────────┐ │
│ ↓ ↓ ↓ │
│ 第5章 第6章 设计模式 │
│ partial 类装饰器 依赖注入 │
│ 偏函数 __init__收参 "配置注入" │
│ 应用模式 __call__执行 而非"运行时解析" │
│ │
│ 实际框架中的三层装饰器模式： │
│ ───────────────────────────── │
│ • Flask: @app.route("/path", methods=["GET"]) │
│ • FastAPI: @app.get("/path", response_model=User) │
│ • Django: @login_required(login_url="/login/") │
│ • pytest: @pytest.mark.parametrize("input,expected", [...]) │
│ • click: @click.option("--name", default="World") │
│ │
│ 装饰器工厂的进阶应用： │
│ ───────────────────────────── │
│ • 中间件注册：@middleware_registry.register("auth") │
│ • 事件监听：@event_handler("user.created") │
│ • 任务调度：@scheduled(interval=60, unit="seconds") │
│ │
└─────────────────────────────────────────────────────────────┘自检清单
- [ ] 我能画出三层嵌套的执行流程图
- [ ] 我能解释为什么需要三层而不是两层
- [ ] 我能写出
repeat、retry类型的装饰器 - [ ] 我能理解
Callable[[Callable[..., T]], Callable[..., T]]的含义 - [ ] 我能用
require_role实现简单的 RBAC - [ ] 我能区分装饰器和装饰器工厂的区别
能力清单
学完本章后，你将能够：
- ✅ 编写带参数的装饰器（三层嵌套）
- ✅ 使用装饰器实现权限验证、重试机制
- ✅ 创建装饰器工厂来生成可配置的装饰器
- ✅ 为装饰器添加附加方法（如
reset、call_count） - ✅ 使用
nonlocal在闭包中维护状态 - ✅ 理解装饰器的完整类型签名体系