将 Python 上下文管理器的迭代器嵌套在“with”中

Question

我有一个返回上下文管理器的迭代器。

我想要一个 pythonic with 语句，它模拟多个嵌套 with 语句的行为，每个语句对应迭代器返回的每个上下文管理器。

有人可能会说，我想要（已弃用的）contextlib.nested 函数的泛化。

Answer 1

来自 文档：

需要支持嵌套可变数量上下文管理器的开发人员可以使用 warnings 模块来抑制 [contextlib.nested< 引发的 DeprecationWarning /code>] 或者使用此函数作为应用程序特定实现的模型。




处理多个上下文管理器的困难在于它们的交互非常重要：例如，您可能__enter__第一个，然后在__enter__第二个中引发异常。这些边缘情况正是导致嵌套被弃用的原因。如果您想支持他们，您必须非常仔细地考虑如何编写代码。您可能希望阅读 PEP-0343 以获取想法。

Answer 2

contextlib.nested 有两个主要问题导致它被弃用。

1. 第一个问题是内部上下文管理器可能会在__init__或__new__期间引发异常，并且这些异常将导致整个with语句在不调用<外部管理器的code>__exit__。
2. 第二个问题更复杂。如果内部管理器之一引发异常，并且外部管理器之一通过在 __exit__ 中返回 True 捕获异常，则仍应执行该块。但在nested的实现中，它只是引发一个RuntimeError而不执行该块。这个问题可能需要完全重写nested。

但只需删除nested 定义中的一个* 就可以解决第一问题！
这改变了行为，使得 nested 不再接受参数列表（这无论如何都没有用，因为 with 已经可以处理它），而只接受迭代器。因此，我将新版本称为“iter_nested”。
然后，用户可以定义一个迭代器，在迭代期间实例化上下文管理器。

带有生成器的示例：

def contexts():
    yield MyContext1()
    yield MyContext2()

with iter_nested(contexts()) as contexts:
    do_stuff(contexts[0])
    do_other_stuff(contexts[1])

原始代码和我更改后的 nested 版本的代码之间的差异如下：

from contextlib import contextmanager

@contextmanager
--- def nested(*managers):
+++ def iter_nested(mgr_iterator):
    --- #comments & deprecation warning
    exits = []
    vars = []
    --- exc = (None, None, None)
    +++ exc = None # Python 3
    try:
        --- for mgr in managers:
        +++ for mgr in mgr_iterator:
            exit = mgr.__exit__
            enter = mgr.__enter__
            vars.append(enter())
            exits.append(exit)
        yield vars
# All of the following is new and fit for Python 3
except Exception as exception:
    exc = exception
    exc_tuple = (type(exc), exc, exc.__traceback__)
else:
    exc_tuple = (None, None, None)
finally:
    while exits:
        exit = exits.pop()
        try:
            if exit(*exc_tuple):
                exc = None
                exc_tuple = (None, None, None)
        except Exception as exception:
            exception.__context__ = exc
            exc = exception
            exc_tuple = (type(exc), exc, exc.__traceback__)
    if exc:
        raise exc

Answer 3

古老的问题，但我偶然发现了它。现在的一种解决方案是使用 contextlib.ExitStack()：

with contextlib.ExitStack() as exit_stack:
    for cm in my_context_manager_iterator:
        exit_stack.enter_context(cm)
    # we are now within those contexts... do more things

它处理其他答案中提到的问题。如果输入一个上下文管理器失败，您已输入的其他上下文管理器将正确退出。

对于异步使用，还有 AsyncExitStack 和 enter_async_context()

Answer 4

此实现 - 或或多或少类似的东西，应该执行后期 contextçlib.nested 过去所做的事情，但如果在进入新上下文时引发异常，则要处理已输入的上下文。

上下文可以作为上下文协议对象或元组传递给它，
其中第一个成员是被调用对象，在托管环境中将使用元组的其余部分作为参数进行调用：

import sys
import traceback


class NestContext(object):
    def __init__(self, *objects):
        self.objects = objects
    def __enter__(self):
        self.contexts = []
        for obj in self.objects:
            if isinstance(obj, tuple):
                try:
                    obj = obj[0](*obj[1:])
                except Exception, error:
                    self.__exit__(type(error), error, sys.exc_info()[2])
                    raise
            try:
                context = obj.__enter__()
            except Exception, error:
                self.__exit__(type(error), error, sys.exc_info()[2])
                raise   
            self.contexts.append(context)
        return self

    def __iter__(self):
        for context in self.contexts:
            yield context

    def __exit__(self, *args):
        for context in reversed(self.contexts):
            try:
                context.__exit__(*args)
            except Exception, error:
                sys.stderr.write(str(error))

if __name__ == "__main__":
    # example uasage

    class PlainContext(object):
        counter  = 0
        def __enter__(self):
            self.counter = self.__class__.counter
            print self.counter
            self.__class__.counter += 1
            return self
        def __exit__(self, *args):
            print "exiting %d" % self.counter

    with NestContext(*((PlainContext,) for i in range(10))) as all_contexts:
        print tuple(all_contexts)