Course Builder个人学习工作台 · 本地模式
Modern Python:面向有经验开发者的回归路线

Iterable、Iterator、Generator 与 collections.abc

目标:能够解释一次 for 循环背后的迭代协议,并根据是否需要索引、长度或单次消费选择 Iterable、Iterator 或 Sequence。

Python 的 for item in source 大致做两件事:先调用 iter(source) 得到 iterator,再反复调用 next(iterator),直到收到 StopIteration。因此:

  • Iterable 能产生 iterator,通常可以重复开始遍历;
  • Iterator 同时是 iterable,并保存“走到哪里”的状态;它通常只能消费一次;
  • Generator 是用 yield 创建 iterator 的便捷方式,函数状态在每次 yield 之间暂停;
  • Sequence 比 Iterable 承诺更多:长度、整数索引和稳定顺序。

函数参数应写它真正需要的最小能力。只遍历一次可接受 Iterable[T];需要 len() 和 items[0] 时应使用 Sequence[T]。这不是追求抽象,而是让签名与实现一致。

完整示例:流式分批

保存为 iteration_lab.py:

from __future__ import annotations

from collections.abc import Iterable, Iterator, Sequence


def batched(items: Iterable[str], size: int) -> Iterator[list[str]]:
    if size <= 0:
        raise ValueError("size must be positive")
    batch: list[str] = []
    for item in items:
        batch.append(item)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch


def first(items: Sequence[str]) -> str:
    if not items:
        raise ValueError("items must not be empty")
    return items[0]


def source() -> Iterator[str]:
    for value in ("a", "b", "c", "d", "e"):
        print(f"produce:{value}")
        yield value


if __name__ == "__main__":
    for group in batched(source(), 2):
        print(f"consume:{group}")
    assert first(("x", "y")) == "x"

运行:

uv run python iteration_lab.py

预期输出:

produce:a
produce:b
consume:['a', 'b']
produce:c
produce:d
consume:['c', 'd']
produce:e
consume:['e']

输出交错说明 generator 是惰性的:batched 没有先把所有输入读完。它只保存当前 batch,因此可以处理很大的输入流。

为什么 isinstance(x, Iterable) 不是万能检测

collections.abc.Iterable 能识别注册类型和实现 __iter__ 的类,但旧式对象还可能只通过 __getitem__ 支持迭代。官方文档指出,判断对象是否实际可迭代的最可靠方式是调用 iter(obj)。通常业务代码无需提前检测;直接迭代并让清楚的 TypeError 暴露错误即可。

引导实验:观察 iterator 被消费

values = (str(i) for i in range(3))
print(list(values))
print(list(values))

第二次得到空 list,因为 generator iterator 已到终点。若函数需要遍历两次,可以:

  • 要求 Sequence;
  • 在边界一次性 items = list(items),并接受内存成本;
  • 重写算法只遍历一次。

哪一种正确取决于数据规模和接口承诺。

独立练习

实现 windowed(items, width),对 a,b,c,d 和宽度 3 依次产生 [a,b,c]、[b,c,d]。允许任何 Iterable,不提前读取全部输入。

参考实现
from collections import deque
from collections.abc import Iterable, Iterator


def windowed(items: Iterable[str], width: int) -> Iterator[list[str]]:
    if width <= 0:
        raise ValueError("width must be positive")
    window: deque[str] = deque(maxlen=width)
    for item in items:
        window.append(item)
        if len(window) == width:
            yield list(window)


assert list(windowed("abcd", 3)) == [["a", "b", "c"], ["b", "c", "d"]]

排错路线

  • generator has no len():签名是否承诺得太弱,或算法能否改成单次遍历?
  • 第二次遍历为空:确认拿到的是 iterator 还是可重新迭代的 collection。
  • generator 内直接 raise StopIteration:用 return 结束;直接抛出会按现代语义转成 RuntimeError。

参考资料