我有一个返回上下文管理器的迭代器。
我想要一个 pythonicwith
语句,它模拟几个嵌套with
语句的行为,一个用于迭代器返回的每个上下文管理器。
有人可能会说,我想要(已弃用)contextlib.nested
功能的概括。
我有一个返回上下文管理器的迭代器。
我想要一个 pythonicwith
语句,它模拟几个嵌套with
语句的行为,一个用于迭代器返回的每个上下文管理器。
有人可能会说,我想要(已弃用)contextlib.nested
功能的概括。
contextlib.nested
有两个主要问题导致它被弃用。
__init__
or期间引发异常__new__
,这些异常会导致整个 with 语句在不调用__exit__
外部管理器的情况下中止。True
in捕获它__exit__
,则仍应执行该块。但是在 的实现中,它只是在不执行块的情况下nested
引发 a 。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
这个实现 - 或者或多或少类似的东西,应该做后期的 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)