PEP 525 异步生成器钩子

PEP 525 提议了一组由新的 sys.{get/set}_asyncgen_hooks() 函数管理的全局线程局部钩子，允许事件循环与垃圾回收器集成，以运行异步生成器的清理。原则上，本提案和 PEP 525 是互补的，就像 with 块和 __del__ 是互补的一样：本提案负责确保在大多数情况下的确定性清理，而 PEP 525 的 GC 钩子清理任何遗漏的部分。但 __aiterclose__ 相比单独的 GC 钩子有许多优点

• GC 钩子语义不属于抽象异步迭代器协议的一部分，而是专门限制在异步生成器具体类型上。如果你有一个使用类实现的异步迭代器，像这样

  class MyAsyncIterator:
      async def __anext__():
          ...
  

  那么你无法在不改变其语义的情况下将其重构为异步生成器，反之亦然。这看起来非常不符合 Python 的风格。（它还留下了这样一个问题：鉴于基于类的异步迭代器面临与异步生成器完全相同的清理问题，它们到底应该怎么做。）另一方面，__aiterclose__ 是在协议级别定义的，因此它对鸭子类型友好，并且适用于所有迭代器，而不仅仅是生成器。

• 希望在非 CPython 实现（如 PyPy）上运行的代码通常不能依赖 GC 进行清理。如果没有 __aiterclose__，几乎可以肯定的是，在 CPython 上开发和测试的开发人员会编写在 PyPy 上使用时会泄漏资源的库。希望针对其他实现的开发人员将不得不采取防御性方法，将每个 for 循环包装在 with 块中，或者仔细审计他们的代码，找出哪些生成器可能包含清理代码，并仅在这些生成器周围添加 with 块。有了 __aiterclose__，编写可移植代码变得简单自然。
• 构建健壮软件的一个重要部分是确保异常始终正确传播而不会丢失。与传统的基于回调的系统相比，async/await 最令人兴奋的一点是，它不再需要手动链接，运行时现在可以承担传播错误的繁重工作，使得编写健壮代码*容易得多*。但是，这个美好的新景象有一个主要缺陷：如果我们依赖 GC 进行生成器清理，那么清理过程中引发的异常就会丢失。因此，再次强调，有了 __aiterclose__，关注这种健壮性的开发人员将不得不采取防御性方法，将每个 for 循环包装在 with 块中，或者仔细审计他们的代码，找出哪些生成器可能包含清理代码。__aiterclose__ 通过在调用者的上下文中执行清理来弥补这个漏洞，因此编写更健壮的代码成为阻力最小的路径。
• WSGI 经验表明，存在重要的基于迭代器的 API，它们需要即时清理，不能依赖 GC，即使在 CPython 中也是如此。例如，考虑一个基于 async/await 和异步迭代器的假设 WSGI 类似 API，其中响应处理器是一个异步生成器，它接受请求头 + 请求体的异步迭代器，并产生响应头 + 响应体。（这实际上是我最初对异步生成器感兴趣的用例，即这不是假设的。）如果我们遵循 WSGI 的要求，即子迭代器必须正确关闭，那么如果没有 __aiterclose__，我们系统中最简约的中间件看起来像这样

  async def noop_middleware(handler, request_header, request_body):
      async with aclosing(handler(request_body, request_body)) as aiter:
          async for response_item in aiter:
              yield response_item
  

  可以说在常规代码中，可以跳过 for 循环周围的 with 块，具体取决于对生成器内部实现的理解程度。但在这里我们必须处理任意的响应处理程序，因此如果没有 __aiterclose__，这种 with 构造是每个中间件的强制性部分。

  __aiterclose__ 允许我们从每个中间件中消除强制性的样板代码和额外的缩进级别

  async def noop_middleware(handler, request_header, request_body):
      async for response_item in handler(request_header, request_body):
          yield response_item
  

因此，__aiterclose__ 方法比 GC 钩子提供了显著的优势。

这留下了一个问题：我们是想要 GC 钩子 + __aiterclose__ 的组合，还是仅仅是 __aiterclose__。由于绝大多数生成器都使用 for 循环或类似的方式进行迭代，因此在 GC 有机会介入之前，__aiterclose__ 处理了大多数情况。GC 钩子提供额外价值的情况是执行手动迭代的代码，例如

agen = fetch_newline_separated_json_from_url(...)
while True:
    document = await type(agen).__anext__(agen)
    if document["id"] == needle:
        break
# doesn't do 'await agen.aclose()'

如果采用 GC 钩子 + __aiterclose__ 方法，这个生成器最终将通过 GC 调用生成器的 __del__ 方法进行清理，然后该方法将使用钩子回调到事件循环中运行清理代码。

如果我们采用无 GC 钩子方法，这个生成器最终将被垃圾回收，其效果如下

• 其 __del__ 方法将发出警告，指出生成器未关闭（类似于现有的“协程从未被等待”警告）。
• 所涉及的底层资源仍将被清理，因为生成器帧仍将被垃圾回收，导致它放弃对所持有的任何文件句柄或套接字的引用，然后这些对象的 __del__ 方法将释放实际的操作系统资源。
• 但是，生成器内部的任何清理代码（例如日志记录、缓冲区刷新）将没有机会运行。

这里的解决方案——正如警告所示——是修复代码，使其调用 __aiterclose__，例如通过使用 with 块

async with aclosing(fetch_newline_separated_json_from_url(...)) as agen:
    while True:
        document = await type(agen).__anext__(agen)
        if document["id"] == needle:
            break

基本上，在这种方法中，规则是如果你想手动实现迭代器协议，那么你有责任实现所有协议，现在这也包括 __(a)iterclose__。

GC 钩子以以下形式增加了非平凡的复杂性：（a）新的全局解释器状态，（b）有点复杂的控制流（例如，异步生成器 GC 总是涉及复活，因此 PEP 442 的细节很重要），以及（c）asyncio 中一个新的公共 API（await loop.shutdown_asyncgens()），用户必须记住在适当的时候调用它。（最后一点尤其削弱了 GC 钩子提供安全备份以保证清理的论点，因为如果 shutdown_asyncgens() 没有正确调用，我*认为*生成器可能会被静默丢弃而没有调用其清理代码；将其与仅 __aiterclose__ 方法进行比较，在最坏的情况下我们至少会打印一个警告。这可能是可修复的。）综合考虑，GC 钩子可能不值得，因为它们只帮助那些想要手动调用 __anext__ 但不想手动调用 __aiterclose__ 的人。但 Yury 在这一点上与我意见相左 :-)。两种选择都可行。

始终注入资源，并在顶层执行所有清理

python-dev 和 python-ideas 上的几位评论者建议，避免这些问题的一种模式是始终从上方传入资源，例如 read_newline_separated_json 应该接受文件对象而不是路径，并在顶层处理清理

def read_newline_separated_json(file_handle):
    for line in file_handle:
        yield json.loads(line)

def read_users(file_handle):
    for document in read_newline_separated_json(file_handle):
        yield User.from_json(document)

with open(path) as file_handle:
    for user in read_users(file_handle):
        ...

这在简单情况下效果很好；在这里它让我们避免了“N+1 个 with 块问题”。但不幸的是，当事情变得更复杂时，它很快就会失效。考虑一下，如果我们的生成器不是从文件读取，而是从流式 HTTP GET 请求读取——同时通过 OAUTH 处理重定向和身份验证。那么我们确实希望套接字在我们的 HTTP 客户端库内部进行管理，而不是在顶层。此外，还有其他情况下，嵌入在生成器内部的 finally 块本身也很重要：数据库事务管理，在清理过程中发出日志信息（WSGI close 的主要动机用例之一），等等。所以这实际上是简单情况下的变通方法，而不是通用解决方案。

__(a)iterclose__ 更复杂的变体

__(a)iterclose__ 的语义在某种程度上受到 with 块的启发，但上下文管理器更强大：__(a)exit__ 可以区分正常退出和异常展开，在发生异常时，它可以检查异常细节并选择性地抑制传播。这里提议的 __(a)iterclose__ 没有这些能力，但可以想象一个具有这些能力的替代设计。

然而，这似乎是不必要的复杂性：经验表明，可迭代对象通常具有 close 方法，甚至具有调用 self.close() 的 __exit__ 方法，但我不知道有任何常见情况会利用 __exit__ 的全部功能。我也想不出任何这会很有用的例子。并且允许迭代器通过吞噬异常来影响流控制似乎会不必要地混淆——如果你处于确实需要这种情况的境地，那么你可能仍然应该使用一个真正的 with 块。

指导原则

一般来说，__(a)iterclose__ 的实现应该

• 幂等，
• 假定在调用 __(a)iterclose__ 后迭代器将不再使用，执行所有适当的清理。特别是，一旦调用了 __(a)iterclose__，那么调用 __(a)next__ 会产生未定义的行为。

通常，任何开始迭代可迭代对象并打算耗尽它的代码，都应该确保最终调用 __(a)iterclose__，无论迭代器是否实际耗尽。

迭代的改变

核心提案是 for 循环行为的改变。给定这段 Python 代码

for VAR in ITERABLE:
    LOOP-BODY
else:
    ELSE-BODY

我们将其分解为等效的

_iter = iter(ITERABLE)
_iterclose = getattr(type(_iter), "__iterclose__", lambda: None)
try:
    traditional-for VAR in _iter:
        LOOP-BODY
    else:
        ELSE-BODY
finally:
    _iterclose(_iter)

这里的“传统 for 语句”是经典 3.5 及更早版本的 for 循环语义的简写。

除了顶层的 for 语句，Python 还包含其他几个消费迭代器的地方。为了保持一致性，这些地方也应该使用与上述等效的语义调用 __iterclose__。这包括

• 推导式中的 for 循环
• * 解包
• 接受并完全消费可迭代对象的函数，例如 list(it)、tuple(it)、itertools.product(it1, it2, ...) 等。

此外，成功耗尽被调用生成器的 yield from 应该作为最后一步调用其 __iterclose__ 方法。（理由：yield from 已经将调用生成器的生命周期链接到被调用生成器；如果调用生成器在 yield from 中途关闭，那么这将自动关闭被调用生成器。）

异步迭代的改变

我们还对异步迭代构造进行了类似的更改，只是新的槽位被称为 __aiterclose__，它是一个异步方法，会被 await。

基本迭代器类型的修改

生成器对象（包括由生成器推导式创建的对象）

• __iterclose__ 调用 self.close()
• __del__ 调用 self.close()（与现在相同），如果生成器未耗尽，还会发出 ResourceWarning。此警告默认隐藏，但对于那些希望确保自己不会无意中依赖 CPython 特定的 GC 语义的人可以启用。

异步生成器对象（包括由异步生成器推导式创建的对象）

• __aiterclose__ 调用 self.aclose()
• 如果尚未调用 aclose，则 __del__ 会发出 RuntimeWarning，因为这可能表明存在潜在错误，类似于“协程从未被等待”警告。

问题：文件对象是否应该实现 __iterclose__ 来关闭文件？一方面，这会使此更改更具破坏性；另一方面，人们非常喜欢编写 for line in open(...): ...，如果习惯了迭代器自行清理，那么文件不这样做可能会变得非常奇怪。

新的便捷函数

operator 模块增加了两个新函数，其语义等同于以下内容

def iterclose(it):
    if not isinstance(it, collections.abc.Iterator):
        raise TypeError("not an iterator")
    if hasattr(type(it), "__iterclose__"):
        type(it).__iterclose__(it)

async def aiterclose(ait):
    if not isinstance(it, collections.abc.AsyncIterator):
        raise TypeError("not an iterator")
    if hasattr(type(ait), "__aiterclose__"):
        await type(ait).__aiterclose__(ait)

itertools 模块获得了一个新的迭代器包装器，可用于选择性地禁用新的 __iterclose__ 行为

# QUESTION: I feel like there might be a better name for this one?
class preserve(iterable):
    def __init__(self, iterable):
        self._it = iter(iterable)

    def __iter__(self):
        return self

    def __next__(self):
        return next(self._it)

    def __iterclose__(self):
        # Swallow __iterclose__ without passing it on
        pass

示例用法（假设文件对象实现了 __iterclose__）

with open(...) as handle:
    # Iterate through the same file twice:
    for line in itertools.preserve(handle):
        ...
    handle.seek(0)
    for line in itertools.preserve(handle):
        ...

@contextlib.contextmanager
def iterclosing(iterable):
    it = iter(iterable)
    try:
        yield preserve(it)
    finally:
        iterclose(it)

迭代器包装器的 __iterclose__ 实现

Python 附带了许多充当其他迭代器包装器的迭代器类型：map、zip、itertools.accumulate、csv.reader 等。这些迭代器应该定义一个 __iterclose__ 方法，该方法依次调用其底层迭代器上的 __iterclose__。例如，map 可以实现为

# Helper function
map_chaining_exceptions(fn, items, last_exc=None):
    for item in items:
        try:
            fn(item)
        except BaseException as new_exc:
            if new_exc.__context__ is None:
                new_exc.__context__ = last_exc
            last_exc = new_exc
    if last_exc is not None:
        raise last_exc

class map:
    def __init__(self, fn, *iterables):
        self._fn = fn
        self._iters = [iter(iterable) for iterable in iterables]

    def __iter__(self):
        return self

    def __next__(self):
        return self._fn(*[next(it) for it in self._iters])

    def __iterclose__(self):
        map_chaining_exceptions(operator.iterclose, self._iters)

def chain(*iterables):
    try:
        while iterables:
            for element in iterables.pop(0):
                yield element
    except BaseException as e:
        def iterclose_iterable(iterable):
            operations.iterclose(iter(iterable))
        map_chaining_exceptions(iterclose_iterable, iterables, last_exc=e)

在某些情况下，这需要一些微妙之处；例如，itertools.tee 不应该在所有克隆迭代器都调用过 __iterclose__ 之前，在底层迭代器上调用 __iterclose__。

示例/理由

所有这些的益处是，我们现在可以编写像下面这样直接的代码

def read_newline_separated_json(path):
    for line in open(path):
        yield json.loads(line)

并确信文件将获得确定性清理，*而无需最终用户付出任何特殊努力*，即使在复杂情况下也是如此。例如，考虑这个简单的管道

list(map(lambda key: key.upper(),
         doc["key"] for doc in read_newline_separated_json(path)))

如果我们的文件包含一个文档，其中 doc["key"] 结果是一个整数，那么将发生以下事件序列

1. key.upper() 引发 AttributeError，该异常从 map 传播出来，并触发 list 内部的隐式 finally 块。
2. list 中的 finally 块调用 map 对象的 __iterclose__()。
3. map.__iterclose__() 调用生成器推导式对象的 __iterclose__()。
4. 这会将一个 GeneratorExit 异常注入到生成器推导式主体中，该主体目前暂停在推导式的 for 循环主体内部。
5. 异常从 for 循环中传播出来，触发 for 循环的隐式 finally 块，该块在表示对 read_newline_separated_json 调用的生成器对象上调用 __iterclose__。
6. 这会在 read_newline_separated_json 的主体中注入一个内部 GeneratorExit 异常，目前暂停在 yield 处。
7. 内部 GeneratorExit 从 for 循环中继续传播，命中生成器函数的边界，并导致 read_newline_separated_json 的 __iterclose__() 方法成功返回。
8. 文件对象已关闭。
9. 内部 GeneratorExit 恢复传播，到达生成器函数的边界，并导致 read_newline_separated_json 的 __iterclose__() 方法成功返回。
10. 控制返回到生成器推导式主体，外部 GeneratorExit 继续传播，允许推导式的 __iterclose__() 成功返回。
11. 其余的 __iterclose__() 调用顺利展开，返回到 list 的主体。
12. 原始的 AttributeError 恢复传播。

（上述细节假设我们实现了 file.__iterclose__；如果不是，则向 read_newline_separated_json 添加一个 with 块，并且基本相同的逻辑也会通过。）

当然，从用户的角度来看，这可以简化为

1. int.upper() 引发 AttributeError 1. 文件对象已关闭。 2. AttributeError 从 list 中传播出来

所以我们已经实现了我们的目标，让这一切“just work”，而用户无需考虑。