Вложите итератор контекстных менеджеров Python в «with»

Question

У меня есть итератор, который возвращает менеджеры контекста.

Мне нужен pythonic with оператор, который имитирует поведение нескольких вложенных with операторов, по одному для каждого менеджера контекста, возвращаемого итератором.

Можно сказать, я хочу обобщение (устаревшей) функции contextlib.nested.

Answer 1

Из документов :

Разработчики, которым необходимо поддерживать вложение переменного числа менеджеров контекста, могут либо использовать модуль warnings для подавления DeprecationWarning, вызванного [contextlib.nested], либо использовать эту функцию в качестве модели для конкретной реализации приложения. .

Сложность при работе с несколькими менеджерами контекста заключается в том, что они взаимодействуют нетривиально: например, вы можете __enter__ первый, а затем вызвать исключение при __enter__ во втором. Подобные крайние случаи - именно то, что вызвало устаревание nested. Если вы хотите поддержать их, вам придется очень тщательно подумать о том, как вы пишете свой код. Вы можете прочитать PEP-0343 для идей.

Answer 2

contextlib.nested имеет две основные проблемы, из-за которых он устарел.

1. Первая проблема заключается в том, что внутренние менеджеры контекста могут вызывать исключения во время __init__ или __new__и эти исключения могут привести к прерыванию оператора целого с вызовом без вызова __exit__ внешнего менеджера.
2. Вторая проблема более сложна.Если один из внутренних менеджеров вызывает исключение и один из внешних менеджеров ловит его, возвращая True в __exit__, блок все равно должен быть выполнен.Но в реализации nested он просто вызывает RuntimeError без выполнения блока.Эта проблема, вероятно, требует полного переписывания nested.

Но решить проблему first можно, просто удалив один * в определении 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

Эта реализация - или что-то более или менее подобное должно делать то, что обычно делал поздний 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)