迭代器与生成器

书架和书签是两回事

列表就像一个可以反复使用、摆放各种值的书架。可迭代对象是任何能够提供书签的东西。迭代器就是那个带有状态的书签：它会记住下一个值在哪里。

topics = ["iterables", "iterators", "generators"]
bookmark = iter(topics)

print(next(bookmark))
print(next(bookmark))
print(list(bookmark))
print(list(bookmark))

输出：

iterables
iterators
['generators']
[]

迭代器只能使用一次，所以第二个 list(bookmark) 是空的。原始列表可以重复使用，因为再次调用 iter(topics) 会创建一个全新的迭代器。文件、字符串、范围以及许多其他对象也都是可迭代对象。

StopIteration 表示“已经走到尽头”

当没有剩余值时，next() 会引发 StopIteration。for 循环会悄悄替你处理这个正常的信号。

bookmark = iter([10, 20])

while True:
    try:
        value = next(bookmark)
    except StopIteration:
        print("done")
        break
    else:
        print(value)

输出：

10
20
done

处理普通任务时，请编写普通的 for 循环。只有当确切的边界很重要时，才显式使用 next()。next(bookmark, "empty") 可以改为返回一个默认值，但你选择的标记不能与真实数据混淆。

不要在生成器函数体内引发 StopIteration。请使用 return 或让函数自然运行到结尾；如果 StopIteration 从生成器中逸出，它会变成 RuntimeError。

yield 是一个暂停按钮

包含 yield 的函数称为生成器函数。调用它会创建一个生成器，但此时函数体还不会运行。第一次取值时，它才会开始运行；每次执行到 yield，它都会交出一个值，并冻结局部状态。

def count_to(limit):
    print("started")
    for number in range(1, limit + 1):
        yield number


numbers = count_to(3)
print("created")
print(next(numbers))
print(list(numbers))

输出：

created
started
1
[2, 3]

列表推导式会立即构建所有内容；而像 (n * n for n in range(5)) 这样的生成器表达式，只会在取值时生成值。如果需要重复遍历、获取长度、使用索引或保存快照，请选择列表。如果只需遍历一次，并且延迟执行或降低峰值内存占用很重要，请选择惰性数据流。惰性执行并不一定意味着速度更快。

谨慎地连接小型数据流

yield from 会依次传递另一个可迭代对象中的每一个值：

def whole_course():
    yield from ["basics", "collections"]
    yield from ["generators", "testing"]


print(list(whole_course()))

输出：

['basics', 'collections', 'generators', 'testing']

惰性执行也会推迟错误的出现。某个错误的数据行可能在远离创建生成器的调用处才失败，因此不仅要测试生成器的创建，还要测试它的取值过程。如果生成器打开了一个文件，那么它暂停时可能会让该文件一直保持打开状态。请在资源的 with 代码块内使用完生成器，或者通过明确的 try/finally 规则来规定资源归谁管理以及如何关闭。只捕获数据流确实能够归类处理的异常。

generator.send(value) 可以实现双向对话，但请把它放在工具箱的边缘位置，谨慎使用。新生成器必须先通过 next(generator) 到达第一个 yield；如果在此之前发送一个非 None 值，就会引发 TypeError。对于普通的数据处理管道，使用函数参数传入数据、使用生成值传出数据，要清晰得多。

构建一条惰性阅读传送带

将这个完整的标准库项目保存为 reading_conveyor.py：

from dataclasses import dataclass
from itertools import islice


@dataclass(frozen=True)
class Reading:
    topic: str
    minutes: int


def joined(*groups):
    for group in groups:
        yield from group


def parse_readings(lines):
    for line_number, line in enumerate(lines, start=1):
        parts = [part.strip() for part in line.split("|")]
        if len(parts) != 2 or not parts[0]:
            raise ValueError(f"line {line_number}: expected topic|minutes")
        try:
            minutes = int(parts[1])
        except ValueError as cause:
            raise ValueError(f"line {line_number}: invalid minutes") from cause
        if minutes <= 0:
            raise ValueError(f"line {line_number}: minutes must be positive")
        yield Reading(parts[0], minutes)


def at_least(readings, minimum):
    for reading in readings:
        if reading.minutes >= minimum:
            yield reading


raw_lines = joined(
    ["Iterators | 25", "Generators | 40"],
    ["Broken line"],
)
selected = at_least(parse_readings(raw_lines), minimum=20)

first_two = list(islice(selected, 2))
assert first_two == [Reading("Iterators", 25), Reading("Generators", 40)]
print("First pull:", first_two)

try:
    next(selected)
except ValueError as error:
    print("Next pull:", error)

运行 python3 reading_conveyor.py：

First pull: [Reading(topic='Iterators', minutes=25), Reading(topic='Generators', minutes=40)]
Next pull: line 3: expected topic|minutes

islice() 只请求两个值，因此一开始不会触碰有问题的第三行。下一次取值时会到达这一行，并引发被推迟的错误。这个断言能够验证顺序和内容，而不必把整个数据源转换成列表。

三个小任务

1. 创建一个迭代器，调用两次 next()，然后将它剩余的内容转换成列表两次。
2. 向 joined() 添加第三组数据，并在运行之前预测合并后的顺序。
3. 将 islice(selected, 2) 改为 islice(selected, 1)，并说明下一次取值会到达哪一行。

准备好学习第 18 章了吗？

• 我能区分可迭代对象和迭代器。
• 我能解释 for 循环背后的 iter()、next() 和 StopIteration。
• 我知道生成器什么时候开始运行，也知道 yield 会暂停什么。
• 我能在可重复使用的列表和只能遍历一次的惰性数据流之间做出选择。
• 我能使用 yield from 连接各个阶段，同时不隐藏资源的所有权。
• 我运行了阅读传送带，并看到了错误只在取值时才出现。

接下来，你将使用装饰器为函数调用添加实用规则，并使用上下文管理器为代码块添加实用规则。