第 2 章装饰器深度实践

2025-07-20

大家好，这里是程序员晚枫，全网同名。

（3 小时直播 / 录播可拆成 2 次 1.5 h）

目标
• 能写：函数装饰器、类装饰器、带参装饰器、可叠加装饰器
• 会调：用 functools.wraps、inspect 保留元数据、查看调用栈
• 敢用：把“重试、缓存、速率限制”变成一行 @ 符号

──────────────────
2.1 装饰器到底是什么（10 min）
一句话：在不修改被装饰对象源码的前提下，动态增加功能。
现场拆解：

def time_it(func):
    def wrapper(*args, **kw):
        import time, functools
        start = time.perf_counter()
        result = func(*args, **kw)
        print(f"{func.__name__} took {time.perf_counter()-start:.4f}s")
        return result
    return wrapper

@time_it
def slow_add(x, y):
    import time; time.sleep(0.5)
    return x + y

问题：slow_add.__name__ 变成 wrapper → 引出 functools.wraps。

──────────────────
2.2 用 wraps 留住身份证（10 min）

from functools import wraps
def time_it(func):
    @wraps(func)
    def wrapper(*args, **kw):
        ...
    return wrapper

对比 help() 输出，强调元数据 = 文档、签名、注解。

──────────────────
2.3 带参装饰器工厂（25 min）
需求：写一个 @retry(times=3, delay=1)

import time, functools, random

def retry(times=3, delay=1, backoff=2):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kw):
            attempt, wait = 1, delay
            while True:
                try:
                    return func(*args, **kw)
                except Exception as e:
                    if attempt >= times:
                        raise
                    jitter = random.uniform(0.5, 1.5)
                    time.sleep(wait * jitter)
                    wait *= backoff
                    attempt += 1
                    print(f"[retry] {func.__name__} -> {attempt}/{times}")
        return wrapper
    return decorator

@retry(times=4, delay=0.2)
def flaky():
    import random
    if random.random() < 0.7:
        raise ValueError("程序员晚枫提醒你：网络有点不稳，正在重试…")
    return "ok"

互动：
• 让学员现场改 print 为 logging
• 如果 backoff=1 会怎样？
• 用 pytest.raises 写测试

──────────────────
2.4 装饰器链顺序与调试（15 min）

1
2
3

@time_it
@retry(times=2)
def task(): ...

执行顺序 = 离函数最近的先执行。
用 inspect.getcallargs 打印完整调用链：

import inspect
def debug_chain(func):
    @wraps(func)
    def wrapper(*args, **kw):
        print(inspect.signature(func).bind(*args, **kw))
        return func(*args, **kw)
    return wrapper

──────────────────
2.5 类装饰器（20 min）
场景：把普通类注册到全局插件表

PLUGINS = {}

def register(name):
    def decorator(cls):
        PLUGINS[name] = cls
        return cls
    return decorator

@register("email")
class EmailNotifier:
    def send(self, msg): ...

对比：元类也能做注册，但类装饰器可读性更高。

──────────────────
2.6 实战：写个可配置 LRU 缓存（45 min）
需求：
• 支持 maxsize 与 ttl
• 线程安全
• 命中率统计

import time, threading, functools

def lru_cache(maxsize=128, ttl=None):
    lock = threading.RLock()
    cache = {}
    hits, misses = 0, 0

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kw):
            nonlocal hits, misses
            key = args + tuple(sorted(kw.items()))
            with lock:
                if key in cache:
                    result, stamp = cache[key]
                    if ttl is None or time.time() - stamp < ttl:
                        hits += 1
                        return result
                misses += 1
            result = func(*args, **kw)
            with lock:
                cache[key] = (result, time.time())
                if len(cache) > maxsize:
                    cache.pop(next(iter(cache)))  # FIFO 简易淘汰
            return result

        def cache_info():
            return {"hits": hits, "misses": misses, "size": len(cache)}

        wrapper.cache_info = cache_info
        return wrapper
    return decorator

@lru_cache(maxsize=2, ttl=5)
def fib(n):
    return n if n < 2 else fib(n-1) + fib(n-2)

if __name__ == "__main__":
    print([fib(i) for i in range(10)])
    print(fib.cache_info())

──────────────────
2.7 调试与单测（20 min）
• pytest-mock 打桩 time.sleep 测试重试
• pytest-benchmark 测缓存命中率
• 用 functools.lru_cache 源码对照学习

──────────────────
2.8 综合案例：给日志监控器加装饰器（30 min）
回到第 1 章的 tail 生成器，要求：

@rate_limit(max_calls=100, period=60)
@retry_network 针对网络推送异常

装饰器顺序必须正确（速率限制→重试）
现场完成骨架：

@rate_limit(max_calls=100, period=60)
@retry(times=3, delay=1)
def push_to_webhook(line):
    import requests
    requests.post("https://...", json={"msg": line})

──────────────────
2.9 小结 & 思维导图（5 min）
函数装饰器 → 带参装饰器 → 类装饰器 → 组合调试 → 实战缓存/重试/限流

──────────────────
2.10 课后作业

必做：把 LRU 缓存改成基于 OrderedDict + TTL 堆，淘汰策略改为 LRU 而不是 FIFO。
选做：写一个 @singleton 类装饰器，支持线程安全的懒汉模式。
挑战：阅读 CPython 3.12 的 @functools.lru_cache 源码，列出 3 个你没想到的优化点。

提交：
• 代码 + 单测 push 到 feat/lesson2 分支
• Pull Request 描述区画一张装饰器执行顺序时序图（可用 mermaid）

──────────────────
附录：常用装饰器速查表
| 功能 | 现成库 | pip 命令 |
|—————|———————————–|———————————|
| 重试 | tenacity | poetry add tenacity |
| 缓存 | functools / cachetools | poetry add cachetools |
| 速率限制 | ratelimit / slowapi | poetry add ratelimit |
| 日志 | logdecorator | poetry add logdecorator |

（第 2 章完）

大家在学习课程中有任何问题，欢迎+微信和我交流👉我的联系方式：微信、读者群、1对1、福利

扫一扫，领红包

美团红包