在一个上下文管理器内处理另一个上下文管理器的实例
在Python中,如何处理在一个上下文管理器内部创建的另一个上下文管理器呢?
举个例子,假设你有一个类A
,它可以作为上下文管理器使用,还有一个类B
,它也可以作为上下文管理器使用。但是类B
的实例需要创建并使用类A
的实例。我看过PEP 343(Python的一份提案文档),这是我想到的解决方案:
class A(object):
def __enter__(self):
# Acquire some resources here
return self
def __exit__(seplf, exception_type, exception, traceback):
# Release the resources and clean up
pass
class B(object):
def __init__(self):
self.a = A()
def __enter__(self):
# Acquire some resources, but also need to "start" our instance of A
self.a.__enter__()
return self
def __exit__(self, exception_type, exception, traceback):
# Release the resources, and make our instance of A clean up as well
self.a.__exit__(exception_type, exception, traceback)
这样做是对的吗?还是说我遗漏了什么需要注意的地方?
4 个回答
1
这里是对kuzzooroo在最受欢迎回答的进一步解释,不过这次是针对Python 3及以上版本:
为了方便,这里是kuzzooroo的原始代码,已经转换成Python 3的格式(主要是在print
语句中加了括号):
import contextlib
@contextlib.contextmanager
def internal_cm():
try:
print("Entering internal_cm")
yield None
print("Exiting cleanly from internal_cm")
finally:
print("Finally internal_cm")
@contextlib.contextmanager
def external_cm():
with internal_cm() as c:
try:
print("In external_cm_f")
yield [c]
print("Exiting cleanly from external_cm_f")
finally:
print("Finally external_cm_f")
if "__main__" == __name__:
with external_cm():
print("Location A")
with external_cm():
print("Location B")
raise Exception("Some exception occurs!!")
这是这个脚本的输出结果:
Entering internal_cm
In external_cm_f
Location A
Exiting cleanly from external_cm_f
Finally external_cm_f
Exiting cleanly from internal_cm
Finally internal_cm
Entering internal_cm
In external_cm_f
Location B
Finally external_cm_f
Finally internal_cm
Traceback (most recent call last):
File "main.py", line 28, in <module>
raise Exception("Some exception occurs!!")
Exception: Some exception occurs!!
2
这里有一个关于手动资源管理的例子,使用了上下文管理器:外层的上下文管理器负责管理内层的。
class Inner:
def __enter__(self):
print("<inner>")
return self
def __exit__(self, exc_type, exc_value, traceback):
print("</inner>")
class Outer:
def __init__(self):
self.inner = Inner()
def __enter__(self):
self.inner.__enter__()
try:
#raise RuntimeError("Suppose we fail here")
print("<outer>")
return self
except Exception as e:
self.inner.__exit__(None, None, None)
raise e
def __exit__(self, exc_type, exc_value, traceback):
print("</outer>")
self.inner.__exit__(exc_type, exc_value, traceback)
使用方法和正常一样:
with Outer() as scope:
#raise RuntimeError("Suppose we fail here")
pass
细心的读者会发现,内层的上下文管理器现在变成了一个没什么用的木偶(因为我们是手动操控它的)。那就这样吧。
9
另外,你可以这样写你的代码:
with A() as a:
with B(a) as b:
# your code here
你也可以试试这个解决方案:
class A:
def __init__(self):
pass
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
pass
class B(A):
def __init__(self):
super().__init__()
def __enter__(self):
super().__enter__()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
super().__exit__(exc_type, exc_val, exc_tb)
在考虑了你的情况后,这可能是一个更好的解决方案:
class Resource:
def __init__(self, dependency=None):
self.dependency = dependency
# your code here
def __enter__(self):
if self.dependency:
self.dependency.__enter__()
# your code here
return self
def __exit__(self, exc_type, exc_val, exc_tb):
# your code here
if self.dependency:
self.dependency.__exit__(exc_type, exc_val, exc_tb)
我不确定下面的实现是否正确,但 __exit__
必须妥善处理异常。对我来说,想象如何在处理异常的同时递归地链接调用有点困难。
class Resource:
def __init__(self, dependency=None):
self.dependency = dependency
self.my_init()
def __enter__(self):
if self.dependency:
self.dependency.__enter__()
return self.my_enter()
def __exit__(self, exc_type, exc_val, exc_tb):
suppress = False
try:
suppress = self.my_exit(exc_type, exc_val, exc_tb)
except:
exc_type, exc_val, exc_tb = sys.exc_info()
if suppress:
exc_type = exc_val = exc_tb = None
if self.dependency:
suppress = self.dependeny.__exit__(exc_type, exc_val, exc_tb)
if not supress:
raise exc_val.with_traceback(exc_tb) from None
return suppress
def my_init(self):
pass
def my_enter(self):
pass
def my_exit(self, exc_type, exc_val, exc_tb):
pass
20
如果你能使用 @contextlib.contextmanager
这个装饰器,你的生活会变得简单很多:
import contextlib
@contextlib.contextmanager
def internal_cm():
try:
print "Entering internal_cm"
yield None
print "Exiting cleanly from internal_cm"
finally:
print "Finally internal_cm"
@contextlib.contextmanager
def external_cm():
with internal_cm() as c:
try:
print "In external_cm_f", c
yield [c]
print "Exiting cleanly from external_cm_f", c
finally:
print "Finally external_cm_f", c
if "__main__" == __name__:
with external_cm():
print "Location A"
print
with external_cm():
print "Location B"
raise Exception("Some exception occurs!!")