Reputation: 6156
I have an iterator that returns context managers.
I want a pythonic with
statement, that emulates the behaviour of several nested with
statements, one for each context manager returned by the iterator.
One could say, I want a generalisation of the (deprecated) contextlib.nested
function.
Upvotes: 5
Views: 3854
Reputation: 373
Ancient question but I stumbled upon it. One solution now is to use 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
It handles the problems mentioned in other answers. If entering one context manager fails, the others you have already entered will be exited properly.
For async usage there is also AsyncExitStack
with enter_async_context()
Upvotes: 1
Reputation: 6156
contextlib.nested
has two major problems that caused it to be deprecated.
__init__
or __new__
, and these exceptions would cause the whole with statement to abort without calling __exit__
of the outer manager.True
in __exit__
, the block should still be executed. But in the implementation of nested
, it just raises a RuntimeError
without executing the block. This problem probably requires a total rewrite of nested
.But it is possible to solve the first problem by just removing one *
in the definition of nested
!
This changes the behaviour such that nested
doesn't accept argument lists anymore (which isn't useful anyway because with
can handle that already) but only an iterator. I therefore call the new version "iter_nested
".
The user can then define an iterator that instantiates the context managers during iteration.
An example with a generator:
def contexts():
yield MyContext1()
yield MyContext2()
with iter_nested(contexts()) as contexts:
do_stuff(contexts[0])
do_other_stuff(contexts[1])
The difference between the codes of the original and my changed version of nested
is here:
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
Upvotes: 1
Reputation: 110516
This implementation - or something more or less like this, should make what the late contextçlib.nested used to do, but taking care of the already entered contexts if an exception is raised when entering a new context.
Contexts can be passed to it either as a context-protocol object, or as a tuple, where the first member is a called object that will be called with the remainder of the tuple as parameters, in a managed environment:
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)
Upvotes: 0
Reputation: 123742
From the docs:
Developers that need to support nesting of a variable number of context managers can either use the
warnings
module to suppress theDeprecationWarning
raised by [contextlib.nested
] or else use this function as a model for an application specific implementation.
The difficult thing about handling multiple context managers is that they interact non-trivially: for example, you might __enter__
the first then raise an exception in __enter__
ing the second. These sort of edge cases are precisely what caused nested
to be deprecated. If you want to support them, you will have to think very carefully about how you write your code. You may wish to read PEP-0343 for ideas.
Upvotes: 3