Python 协议进阶：迭代、生成器、对象与资源

前置：函数与模块、类、异常。教学与验证使用标准库 CPython 3.12.14/3.14.4；不把3.14特有API混入3.12主线。完整可运行 examples/python/pipeline-lab/protocols.py：

"""Run directly to check object and generator protocols."""
from dataclasses import dataclass, field
from functools import wraps
from contextlib import contextmanager


@dataclass
class Notebook:
    pages: list[str] = field(default_factory=list)


class Positive:
    def __set_name__(self, owner, name):
        self.storage = '_' + name

    def __get__(self, instance, owner=None):
        if instance is None:
            return self
        return getattr(instance, self.storage)

    def __set__(self, instance, value):
        if value <= 0:
            raise ValueError('positive required')
        setattr(instance, self.storage, value)


class Limit:
    size = Positive()
    def __init__(self, size):
        self.size = size


def logged(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        return function(*args, **kwargs)
    return wrapper


def accumulator():
    total = 0
    while (value := (yield total)) is not None:
        total += value
    return total


@contextmanager
def managed(events):
    events.append('open')
    try:
        yield events
    finally:
        events.append('close')


def main():
    a, b = Notebook(), Notebook()
    a.pages.append('one')
    assert b.pages == []
    shallow = a.pages.copy()
    assert shallow == a.pages and shallow is not a.pages
    limit = Limit(2)
    limit.__dict__['size'] = -1
    assert limit.size == 2  # data descriptor has precedence over instance __dict__
    try:
        limit.size = 0
    except ValueError:
        pass
    else:
        raise AssertionError('descriptor did not reject')
    values = [1, 2]
    it = iter(values)
    assert iter(it) is it
    assert list(it) == [1, 2] and list(it) == []
    generator = accumulator()
    assert next(generator) == 0
    assert generator.send(3) == 3
    try:
        generator.send(None)
    except StopIteration as done:
        assert done.value == 3
    events = []
    try:
        with managed(events):
            raise ValueError('body failed')
    except ValueError:
        pass
    assert events == ['open', 'close']
    assert logged(len)([1, 2]) == 2
    print('protocols ok')


if __name__ == '__main__':
    main()

在该目录 PowerShell 执行 python protocols.py，Bash 执行 python3 protocols.py。所有断言成立并输出 protocols ok。下面按机制解释断言为什么成立。

本轮复核仅静态阅读正文、源码与测试规格，未执行示例或测试；文中原有实测版本记录属于此前批次，本轮不复用为修改后的运行证明。命令与断言结果均为读者后续验证预期。

可迭代对象与一次性迭代器

Python迭代器耗尽后不会自动从头开始，再次next仍抛StopIteration；要重新遍历可迭代容器，应重新取得迭代器，而不是复用已经耗尽的游标。

可迭代对象支持 iter(x)，返回实现 next 的迭代器。迭代器的 iter 返回自己，next 在结束时抛 StopIteration。列表通常每次 iter 得到新游标；同一个迭代器消费后不会自动重置。for 循环反复 next，到 StopIteration 正常结束，不把其他异常当成结束。给调用方 Iterable 契约比要求 list 更灵活，但不要悄悄遍历两次；需要重用时显式物化并承认内存成本。

records(lines) 在流水线中逐行验证，调用生成器函数不会立刻读输入，next 才执行到 yield。配套测试的第三行是坏数据，因此推进到第三行才抛带行号的 ValueError，前两行仍可读取；这不是任意输入固定在第三行失败。文件由调用方 with 负责关闭，生成器不假定拥有传入文件。提前停止遍历时必须考虑资源归属；不要把清理寄托在垃圾回收时间上。

生成器的状态与返回

yield 挂起帧并保存局部变量；next 推进时向 yield 表达式送 None，send(value) 让挂起点接收值，首次推进必须 next 或 send(None)。示例 accumulator 先产出0，再 send(3) 产出3，send(None) 命中结束条件，return 的值进入 StopIteration.value。yield from 不仅转发值，还转发 send/throw/close 协议，并取得子生成器的返回值。

不要在生成器内部用 raise StopIteration 表示正常结束：它会按生成器协议变成 RuntimeError，应 return。close() 在挂起点引入 GeneratorExit，finally 可清理；清理中继续 yield 会出错。生成器不是自动异步，也不保证更快；它减少一次性中间列表，仍会保留当前帧引用的大对象。

名称、相等、哈希与数据模型

赋值复制引用，浅拷贝复制外层容器，内部对象仍共享；递归结构和外部资源不宜盲目 deepcopy。is 比较身份，== 调用值比较；小整数/字符串驻留不能作为业务保证。自定义相等应返回适当结果或 NotImplemented；相等对象必须有相同哈希，可变字段参与哈希后再修改会破坏字典/set 查找不变量。

流水线 Record 用 @dataclass(frozen=True, slots=True) 表达可哈希值。frozen 禁止普通属性重绑定，但不让嵌套列表自动不可变；这里字段都是 str/int，适合值语义。slots 限定属性存储并可能省内存，不是安全隔离，也不能推导所有继承场景布局。Notebook 的 default_factory=list 为每个实例创建独立列表；直接类属性列表会共享，函数默认列表也只创建一次。

属性查找会涉及描述符与类继承的MRO。示例 Positive 同时有 get/set，是数据描述符，优先于实例字典同名项；因此向 __dict__['size'] 写-1不能覆盖它的验证。普通函数作为描述符绑定 self；property 也使用描述符机制。super() 沿MRO找下一个实现，不等于硬编码“父类”；多继承协作时签名与super调用必须一致。类装饰器、元类和 __getattribute__ 能改变行为，日常设计先用普通函数、组合和property，避免隐藏控制流。

装饰器、with 与异常

装饰器在函数定义时包装函数，wrapper 的 *args/**kwargs 转发调用，wraps 保留名称等元信息，不自动保持所有类型契约。装饰器层叠按从内到外应用；闭包捕获循环变量需考虑延迟绑定，用显式参数或工厂函数固定每次值。

with 调用上下文管理协议，退出时即使主体出错也执行清理；__exit__ 返回真会抑制异常。示例 contextmanager 的 try/finally 记录 open/close，并让 ValueError 继续传播。它要求 yield 一次；不能把主体错误吞掉然后假装成功。异步资源用 async with/异步退出协议，不把同步阻塞文件I/O自动当作异步。

练习

① it=iter([1,2]); sum(it); sum(it) 第二次值是什么？② dataclass frozen 中若放列表，列表能改吗、对象能哈希吗？③ 让生成器只读前一行就停，谁关闭文件？④ property/描述符验证可否替代输入边界验证？

参考解法

① 0，第一次消费完游标。② 列表内容仍能改，默认生成的hash遇到列表会TypeError；真正值语义要采用不可变字段。③ 外层 with 的拥有者关闭；若生成器自己打开文件，要显式close/上下文管理，不能期待所有解释器立刻回收。④ 不能；协议控制对象访问路径，反序列化、构造与业务输入仍需按契约验证，错误应携带位置。

资料

2026-10-17核对官方中文数据模型、英文迭代器、生成器表达式及 yield、dataclasses、contextlib。国内外检索交叉比较术语，机制以对应版本官方文档为依据。