前言

装饰器和生成器是 Python 进阶的两张门票。前者是 AOP（切面编程）的 Python 实现，后者是"惰性计算"的根基。这篇把原理掰开揉碎。

一、装饰器：从零推导

本质：装饰器就是一个"接收函数、返回函数"的函数。

def my_decorator(func):
    def wrapper(*args, **kwargs):
        print(">>> 调用前")
        result = func(*args, **kwargs)      # 调用原函数
        print(">>> 调用后")
        return result
    return wrapper

# 手动版
def hello():
    print("hello")

hello = my_decorator(hello)     # 手动包一层
hello()                         # 打印前后缀 + hello

# 语法糖版（完全等价）
@my_decorator
def hello():
    print("hello")

@ 语法糖只是 hello = my_decorator(hello) 的缩写——想通这一点，一切装饰器都能自己写出来。

functools.wraps：必写的礼仪

from functools import wraps

def my_decorator(func):
    @wraps(func)                     # 把原函数的 __name__/docstring 拷给 wrapper
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def hello():
    """say hello"""

print(hello.__name__)    # 有 @wraps: 'hello'; 没有: 'wrapper'（文档/调试全乱）

二、常用装饰器实战

计时器

import time, functools

def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        cost = time.perf_counter() - start
        print(f"{func.__name__} 耗时 {cost:.3f}s")
        return result
    return wrapper

@timer
def train_model():
    time.sleep(1.2)

重试器（带参数的装饰器 = 三层嵌套）

def retry(times=3, delay=1):
    def decorator(func):                       # 第2层: 收函数
        @functools.wraps(func)
        def wrapper(*args, **kwargs):           # 第3层: 收调用参数
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == times:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(times=5, delay=2)        # 先调 retry(...) 得到真正的装饰器
def fetch(url): ...

记忆化：@functools.lru_cache 直接用：

@functools.lru_cache(maxsize=1024)
def fib(n):
    return n if n < 2 else fib(n-1) + fib(n-2)
fib(200)        # 瞬间出结果

注册器（框架的路由就是这原理）

routes = {}

def route(path):
    def decorator(func):
        routes[path] = func             # 注册进表, 不改函数行为
        return func
    return decorator

@route("/index")
def index(): return "home"

@route("/api/health")
def health(): return "ok"

print(routes)      # {'/index': <function index>, ...}

三、生成器：yield 的魔法

普通函数一次性返回；生成器函数"暂停执行、分次产出"：

def counter(n):
    i = 0
    while i < n:
        yield i               # 到这里暂停, 把 i 交给调用方
        i += 1                # 下次要下一个值时, 从这里继续

g = counter(3)
print(g)          # <generator object counter at 0x...>
next(g)           # 0（执行到 yield 暂停）
next(g)           # 1（从上次暂停处继续）
next(g)           # 2
next(g)           # StopIteration（耗尽）

for x in counter(5): ...      # for 循环自动处理 StopIteration

为什么用生成器：内存

# 一次性物化 1000 万个数: 瞬间吃掉几百 MB
squares_list = [x * x for x in range(10_000_000)]
sum(squares_list)

# 生成器表达式: 几乎不占内存, 边产边算
squares_gen = (x * x for x in range(10_000_000))
sum(squares_gen)

注意 [] 与 () 的区别：列表推导式 vs 生成器表达式。

逐行读大文件（生成器经典场景）

def error_lines(path):
    with open(path, encoding="utf-8") as f:
        for line in f:              # 文件对象本身就是生成器
            if "ERROR" in line:
                yield line.strip()

# 100GB 的日志也稳: 内存里永远只有一行
for line in error_lines("/var/log/huge.log"):
    handle(line)

生成器管道（Unix 管道思想）

def read(path):
    with open(path) as f:
        yield from f                    # yield from: 委托给子生成器

def strip(lines):
    for l in lines:
        yield l.strip()

def only_error(lines):
    for l in lines:
        if "ERROR" in l:
            yield l

pipeline = only_error(strip(read("app.log")))
for line in pipeline:
    print(line)

每层只做一件事，数据像水流过管道，全程 O(1) 内存。

四、进阶：send 与协程雏形

def averager():
    total = count = 0
    average = None
    while True:
        value = yield average       # 产出平均值, 并接收 send 进来的新值
        if value is None:
            break
        total += value
        count += 1
        average = total / count

avg = averager()
next(avg)               # 启动（执行到第一个 yield）
avg.send(10)            # 10.0
avg.send(20)            # 15.0
avg.send(30)            # 20.0

yield 既能产出又能接收——这就是早期协程的原型，也是 async/await 的思想前身（现代异步编程后面专篇讲）。

五、使用决策

需求 工具
通用横切逻辑（计时/日志/重试/鉴权） 装饰器
函数结果缓存 @lru_cache
大数据流式处理 生成器
惰性无限序列 生成器
组装多步处理管线 生成器管道

小结

  • 装饰器 = 接收函数返回函数；@x 就是 f = x(f)
  • 写装饰器必带 functools.wraps
  • 带参数的装饰器要多包一层
  • 生成器 = 可暂停的函数；yield from 委托子生成器
  • 大数据永远想生成器：内存 O(1)

本文是「Python」系列第 7 篇。