🐍装饰器进阶 — 类装饰器、装饰器链与实战
🕐 预计用时：2-3 小时 ｜ 🎯 目标：掌握类装饰器、装饰器链、实战场景（日志/权限/缓存/重试）
📖 今日目录
1. 装饰器的执行顺序
def decorator_a(func):
print(f"🔴 装饰器 A 被加载")
def wrapper():
print(" → A 前置")
result = func()
print(" → A 后置")
return result
return wrapper
def decorator_b(func):
print(f"🔵 装饰器 B 被加载")
def wrapper():
print(" → B 前置")
result = func()
print(" → B 后置")
return result
return wrapper
@decorator_a
@decorator_b
def say_hello():
print(" → 原函数执行")
return "Hello"
# 加载时的输出（定义时就执行！）:
# 🔵 装饰器 B 被加载
# 🔴 装饰器 A 被加载
# 调用时的输出:
say_hello()
# → A 前置
# → B 前置
# → 原函数执行
# → B 后置
# → A 后置
💡 执行顺序口诀：
加载：从下往上（先 B 后 A）
调用：从上往下（A → B → 原函数 → B → A）
像洋葱一样，一层一层剥进去，再一层一层包回来
2. 装饰器链 — 多个装饰器叠加
from functools import wraps
def log(func):
"""日志装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
print(f"📝 [{func.__name__}] 开始执行")
result = func(*args, **kwargs)
print(f"📝 [{func.__name__}] 执行完毕")
return result
return wrapper
def validate_positive(func):
"""验证参数为正数"""
@wraps(func)
def wrapper(*args, **kwargs):
for arg in args:
if isinstance(arg, (int, float)) and arg < 0:
raise ValueError(f"参数不能为负数: {arg}")
return func(*args, **kwargs)
return wrapper
def cache_result(func):
"""简单缓存"""
cache = {}
@wraps(func)
def wrapper(*args):
if args not in cache:
cache[args] = func(*args)
print(f"💾 缓存: {args} → {cache[args]}")
return cache[args]
return wrapper
# 组合使用
@log
@validate_positive
@cache_result
def power(base, exp):
return base ** exp
print(power(2, 10))
# 📝 [power] 开始执行
# 💾 缓存: (2, 10) → 1024
# 📝 [power] 执行完毕
# 1024
print(power(2, 10)) # 第二次直接命中缓存
# 📝 [power] 开始执行
# 📝 [power] 执行完毕
# 1024
3. 类装饰器 — 装饰类
装饰器不仅能装饰函数，还能装饰类——修改类的行为。
from functools import wraps
def add_repr(cls):
"""给类自动添加 __repr__ 方法"""
def __repr__(self):
attrs = ", ".join(f"{k}={v!r}" for k, v in self.__dict__.items())
return f"{cls.__name__}({attrs})"
cls.__repr__ = __repr__
return cls
@add_repr
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
p = Point(3, 4)
print(p) # Point(x=3, y=4) ← 自动生成的 repr！
# 类装饰器：自动注册子类
registry = {}
def register(name):
def decorator(cls):
registry[name] = cls
return cls
return decorator
@register("dog")
class Dog:
def speak(self):
return "汪汪!"
@register("cat")
class Cat:
def speak(self):
return "喵~"
# 通过名字创建对象
animal = registry["dog"]()
print(animal.speak()) # 汪汪!
print(registry) # {'dog': <class 'Dog'>, 'cat': <class 'Cat'>}
4. 装饰器与类方法
from functools import wraps
def log_method(func):
"""装饰类方法"""
@wraps(func)
def wrapper(self, *args, **kwargs):
print(f"📝 [{self.__class__.__name__}.{func.__name__}] 调用")
return func(self, *args, **kwargs)
return wrapper
class Calculator:
@log_method
def add(self, a, b):
return a + b
@log_method
def multiply(self, a, b):
return a * b
calc = Calculator()
calc.add(3, 5)
# 📝 [Calculator.add] 调用
calc.multiply(4, 6)
# 📝 [Calculator.multiply] 调用
# 装饰 @classmethod 和 @staticmethod
from functools import wraps
def log_func(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"📞 调用 {func.__name__}")
return func(*args, **kwargs)
return wrapper
class MyClass:
@classmethod
@log_func # 注意顺序：@classmethod 在外面
def create(cls):
return cls()
@staticmethod
@log_func
def helper():
return "helper result"
obj = MyClass.create()
# 📞 调用 create
⚠️ 装饰器顺序很重要！
@classmethod / @staticmethod 放在最外面（最后加载）
自定义装饰器放在里面（先执行）
5. 实战：日志装饰器
import time
import logging
from functools import wraps
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)
def log_call(level=logging.INFO, message=""):
"""高级日志装饰器
Args:
level: 日志级别
message: 额外消息
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
func_name = func.__name__
extra = f" ({message})" if message else ""
logger.log(level, f"▶ {func_name}{extra} 开始 | args={args}, kwargs={kwargs}")
start = time.time()
try:
result = func(*args, **kwargs)
elapsed = time.time() - start
logger.log(level, f"✅ {func_name} 成功 | 耗时 {elapsed:.4f}s | 返回 {result!r}")
return result
except Exception as e:
elapsed = time.time() - start
logger.error(f"❌ {func_name} 失败 | 耗时 {elapsed:.4f}s | 错误: {e}")
raise
return wrapper
return decorator
@log_call(level=logging.INFO, message="用户操作")
def process_order(order_id, amount):
"""处理订单"""
if amount <= 0:
raise ValueError("金额必须为正数")
return {"order_id": order_id, "status": "success"}
process_order("ORD001", 99.9)
# 2024-01-15 10:30:15 [INFO] ▶ process_order (用户操作) 开始 | args=('ORD001', 99.9), kwargs={}
# 2024-01-15 10:30:15 [INFO] ✅ process_order 成功 | 耗时 0.0001s | 返回 {'order_id': 'ORD001', 'status': 'success'}
6. 实战：权限校验装饰器
from functools import wraps
# 模拟当前用户（实际项目中从 session/token 获取）
current_user = {"name": "张三", "role": "editor"}
def require_role(*roles):
"""要求用户具有指定角色之一"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
user_role = current_user.get("role")
if user_role not in roles:
print(f"🚫 拒绝访问: {current_user['name']} ({user_role})")
print(f" 需要角色: {', '.join(roles)}")
raise PermissionError(f"权限不足: 需要 {roles}")
print(f"✅ 允许访问: {current_user['name']} ({user_role})")
return func(*args, **kwargs)
return wrapper
return decorator
def require_login(func):
"""要求用户已登录"""
@wraps(func)
def wrapper(*args, **kwargs):
if "name" not in current_user:
print("🚫 请先登录")
raise PermissionError("未登录")
return func(*args, **kwargs)
return wrapper
@require_login
@require_role("admin", "editor")
def publish_article(title, content):
"""发布文章"""
print(f"📰 文章已发布: {title}")
publish_article("Python 装饰器", "装饰器很强大...")
# ✅ 允许访问: 张三 (editor)
# 📰 文章已发布: Python 装饰器
7. 实战：重试装饰器
import time
from functools import wraps
def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
"""自动重试装饰器
Args:
max_attempts: 最大重试次数
delay: 重试间隔（秒）
exceptions: 需要重试的异常类型
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
print(f"⚠️ {func.__name__} 第 {attempt} 次失败: {e}")
if attempt < max_attempts:
print(f" {delay}秒后重试...")
time.sleep(delay)
print(f"❌ {func.__name__} 全部 {max_attempts} 次尝试失败")
raise last_exception
return wrapper
return decorator
# 使用
@retry(max_attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
"""模拟不稳定的数据请求"""
import random
if random.random() < 0.7: # 70% 概率失败
raise ConnectionError(f"连接 {url} 失败")
return {"data": "success"}
try:
result = fetch_data("https://api.example.com")
print(f"结果: {result}")
except ConnectionError:
print("最终失败")
8. 实战：限流装饰器
import time
from functools import wraps
def rate_limit(calls_per_second=1):
"""限流装饰器：限制函数调用频率"""
min_interval = 1.0 / calls_per_second
last_call_time = [0.0] # 用列表存储（闭包可修改）
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_call_time[0]
if elapsed < min_interval:
wait_time = min_interval - elapsed
print(f"⏳ 限流: 等待 {wait_time:.2f}秒")
time.sleep(wait_time)
last_call_time[0] = time.time()
return func(*args, **kwargs)
return wrapper
return decorator
@rate_limit(calls_per_second=2) # 每秒最多 2 次
def api_call(endpoint):
print(f"🌐 请求: {endpoint}")
for i in range(5):
api_call(f"/api/data/{i}")
# 🌐 请求: /api/data/0
# ⏳ 限流: 等待 0.50秒
# 🌐 请求: /api/data/1
# ⏳ 限流: 等待 0.50秒
# ...
9. 实战：类型检查装饰器
from functools import wraps
import inspect
def type_check(func):
"""根据类型注解检查参数类型"""
sig = inspect.signature(func)
annotations = func.__annotations__
@wraps(func)
def wrapper(*args, **kwargs):
# 绑定参数
bound = sig.bind(*args, **kwargs)
bound.apply_defaults()
# 检查每个参数
for name, value in bound.arguments.items():
if name in annotations:
expected_type = annotations[name]
if not isinstance(value, expected_type):
raise TypeError(
f"参数 '{name}' 期望类型 {expected_type.__name__}, "
f"实际得到 {type(value).__name__}: {value!r}"
)
result = func(*args, **kwargs)
# 检查返回值
if 'return' in annotations:
expected = annotations['return']
if not isinstance(result, expected):
raise TypeError(
f"返回值期望类型 {expected.__name__}, "
f"实际得到 {type(result).__name__}"
)
return result
return wrapper
@type_check
def add(a: int, b: int) -> int:
return a + b
@type_check
def greet(name: str, times: int = 1) -> str:
return (f"你好 {name}! " * times).strip()
print(add(3, 5)) # 8
print(greet("张三", 2)) # 你好 张三! 你好 张三!
try:
add("hello", 5) # ❌ TypeError
except TypeError as e:
print(f"错误: {e}") # 参数 'a' 期望类型 int, 实际得到 str: 'hello'
10. 装饰器最佳实践
✅ 装饰器 DO：
- ✅ 必须加
@wraps(func) — 保留原函数的 __name__、__doc__、__module__ - ✅ 用
*args, **kwargs — 让 wrapper 接受任意参数 - ✅ 返回原函数的返回值 —
return func(*args, **kwargs) - ✅ 处理异常 — try/except 或者 re-raise
❌ 装饰器 DON'T：
- ❌ 不要在装饰器里做耗时操作（装饰器在导入时执行）
万能模板
from functools import wraps
def my_decorator(func):
"""装饰器说明"""
@wraps(func)
def wrapper(*args, **kwargs):
# 前置逻辑
try:
result = func(*args, **kwargs)
# 后置逻辑
return result
except Exception as e:
# 异常处理
raise
return wrapper
11. 今日小结
| | |
|---|
| | @A @B @C |
| | @add_repr |
| | @log_call |
| | @require_role("admin") |
| | @retry(times=3) |
| | @rate_limit(rps=2) |
| | @type_check |
核心要点
- ✅ 装饰器链加载顺序：从下往上；调用顺序：从上往下
- ✅
@classmethod/@staticmethod 放最外面
🎯 练习建议：
1. 写一个 @deprecated(reason="请使用 new_func") 装饰器，调用时打印警告
2. 写一个 @timeout(seconds=5) 装饰器，超时抛出 TimeoutError
3. 把今天的所有装饰器整合成一个 utils.py 工具模块
📚 Day35 完成！明天学习多线程 — 并发执行的入门
