迭代器、生成器与装饰器

迭代协议让算法可以逐项消费数据而不关心数据来自列表、文件还是数据库。生成器把“产生下一个值”的状态保存起来，装饰器则在不改变调用方式的前提下组合函数行为。

一、学习目标

区分可迭代对象与迭代器
编写惰性生成器管道
理解 yield、生成器表达式和耗尽
使用 functools.wraps 编写装饰器
避免在数据流中意外重复消费

二、迭代协议

tasks = ["阅读", "测试"]
iterator = iter(tasks)
print(next(iterator))
print(next(iterator))

可迭代对象实现 __iter__，能产生迭代器；迭代器还实现 __next__，没有元素时抛出 StopIteration。for 循环负责处理这些细节。

from collections.abc import Iterable


def print_tasks(titles: Iterable[str]) -> None:
    for title in titles:
        print(title)

参数声明为 Iterable，调用方就可以传列表、元组、生成器等，不必强制构造列表。

三、生成器函数

from collections.abc import Iterable, Iterator


def todo_titles(tasks: Iterable[dict[str, object]]) -> Iterator[str]:
    for task in tasks:
        if task["status"] == "todo":
            yield str(task["title"])

调用生成器函数不会立刻执行函数体，而是返回生成器对象。每次 next 执行到下一个 yield，局部状态会被保留。

for title in todo_titles(tasks):
    print(title)

四、惰性管道

逐行读取大文件：

import json
from collections.abc import Iterator
from pathlib import Path


def read_json_lines(path: Path) -> Iterator[dict[str, object]]:
    with path.open(encoding="utf-8") as file:
        for number, line in enumerate(file, start=1):
            if not line.strip():
                continue
            value = json.loads(line)
            if not isinstance(value, dict):
                raise ValueError(f"第 {number} 行不是对象")
            yield value

过滤与转换：

todo = (
    str(task["title"])
    for task in read_json_lines(Path("tasks.jsonl"))
    if task.get("status") == "todo"
)

整个流程不会一次把文件加载进内存。

五、生成器只能消费一次

values = (number * 2 for number in range(3))
print(list(values))  # [0, 2, 4]
print(list(values))  # []

如果数据需要多次遍历，可以：

重新创建生成器；
明确物化为列表；
调整算法，让统计和写入在一次遍历中完成。

不要把生成器作为长期共享状态传给多个消费者。

六、yield from

from collections.abc import Iterable, Iterator


def flatten(groups: Iterable[Iterable[str]]) -> Iterator[str]:
    for group in groups:
        yield from group

yield from 把子迭代器产生的值转交给调用方，也能正确传递生成器协议中的发送、返回和异常行为。

七、装饰器基础

from collections.abc import Callable
from functools import wraps
from time import perf_counter
from typing import ParamSpec, TypeVar

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


def timed(function: Callable[P, R]) -> Callable[P, R]:
    @wraps(function)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        started = perf_counter()
        try:
            return function(*args, **kwargs)
        finally:
            elapsed = perf_counter() - started
            print(f"{function.__name__}: {elapsed:.3f}s")

    return wrapper

使用：

@timed
def import_tasks(path: str) -> int:
    return 42

wraps 保留原函数的名称、文档和 __wrapped__，有利于调试、测试和框架检查。

八、带参数装饰器

def retry(attempts: int):
    if attempts < 1:
        raise ValueError("attempts 必须大于 0")

    def decorate(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            last_error = None
            for _ in range(attempts):
                try:
                    return function(*args, **kwargs)
                except OSError as exc:
                    last_error = exc
            assert last_error is not None
            raise last_error

        return wrapper

    return decorate

重试只适合暂时性错误，并且操作必须允许重复执行。创建任务等非幂等写入不能无条件重试。

九、什么时候不用装饰器

装饰器会让调用栈增加一层隐式行为。下面情况更适合显式函数或上下文管理器：

行为只用于一个调用点；
需要在函数中间控制开始和结束；
装饰器改变了返回类型或异常语义；
调用方必须明确知道副作用。

十、练习与自测

编写生成器，逐项产生未完成任务的 ID。
使用 JSON Lines 保存十万条任务，比较生成器与 read() 的内存特点。
为只读函数编写限定 OSError 的重试装饰器，并测试最终异常。

自测：

可迭代对象和迭代器有何区别？
为什么生成器第二次遍历可能为空？
functools.wraps 保留了哪些重要信息？

十一、官方资料

上一篇：异常、上下文管理与资源安全｜下一篇：类型注解、dataclass 与结构化数据