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。