具有`上下文与发生器/协程/任务的Python'
我想尝试使用python with块来在该块中应用修饰符。 但我不确定在协同程序的存在下是否可以做到这一点。
例如，假设我有一个WithContext对象，它临时推入一个堆栈，如下所示：
class WithContext:
stack = []
def __init__(self, val):
self.val = val
def __enter__(self):
WithContext.stack.append(self.val)
def __exit__(self, exc_type, exc_val, exc_tb):
WithContext.stack.pop()
def do_scoped_contextual_thing():
print(WithContext.stack[-1])
（显然，堆栈成员必须是线程本地的，但现在忽略。）
然后这个代码：
with WithContext("a"):
do_scoped_contextual_thing()
with WithContext("b"):
do_scoped_contextual_thing()
with WithContext("c"):
do_scoped_contextual_thing()
将打印：
a
b
c
但是现在假设我有一个协同工作的情况：
def coroutine():
with WithContext("inside"):
yield 1
do_scoped_contextual_thing()
yield 2
with WithContext("outside"):
for e in coroutine():
do_scoped_contextual_thing()
print("got " + str(e))
我想要这个代码输出：
outside
got 1
inside
outside
got 2
但实际上它会输出：
inside
got 1
inside
inside
got 2
外部变成了内部，因为协程内部的__enter__在堆栈顶部放置了一个值，并且直到协程结束才调用__exit__ （而不是在您__exit__和退出协程时不断地输入和退出）。
有没有办法解决这个问题？ 是否存在“协同本地”变量？
我对此感觉不好，但是我修改了测试代码，重新输入了几次协程。 类似于@ CraigGidney的解决方案，它使用inspect模块访问和缓存创建WithContext对象的调用栈（又名“范围”）的信息。
然后，我基本上搜索堆栈寻找一个缓存值，并使用id函数来尝试并避免保持对实际框架对象的引用。
import inspect
class WithContext:
stack = []
frame_to_stack = {}
def __init__(self, val):
self.val = val
def __enter__(self):
stk = inspect.stack(context=3)
caller_id = id(stk[1].frame)
WithContext.frame_to_stack[caller_id] = len(WithContext.stack)
WithContext.stack.append( (caller_id, self.val))
def __exit__(self, exc_type, exc_val, exc_tb):
wc = WithContext.stack.pop()
del WithContext.frame_to_stack[wc[0]]
def do_scoped_contextual_thing():
stack = inspect.stack(context=0)
f2s = WithContext.frame_to_stack
for f in stack:
wcx = f2s.get(id(f.frame))
if wcx is not None:
break
else:
raise ValueError("No context object in scope.")
print(WithContext.stack[wcx][1])
def coroutine():
with WithContext("inside"):
for n in range(3):
yield 1
do_scoped_contextual_thing()
yield 2
with WithContext("outside"):
for e in coroutine():
do_scoped_contextual_thing()
print("got " + str(e))
一种可能的半破解“解决方案”是将上下文与堆栈帧的位置相关联，并在查找上下文时检查该位置。
class WithContext:
_stacks = defaultdict(list)
def __init__(self, val):
self.val = val
def __enter__(self):
_, file, _, method, _, _ = inspect.stack()[1]
WithContext._stacks[(file, method)].append(self.val)
def __exit__(self, exc_type, exc_val, exc_tb):
_, file, _, method, _, _ = inspect.stack()[1]
WithContext._stacks[(file, method)].pop()
@staticmethod
def get_context():
for frame in inspect.stack()[1:]:
_, file, _, method, _, _ = frame
r = WithContext._stacks[(file, method)]
if r:
return r[-1]
raise ValueError("no context")
请注意，不断查找堆栈框架比仅传递值更昂贵，并且您可能不想告诉人们您已经写了这个。
请注意，这仍然会在更复杂的情况下破裂。
例如：
我有同样的问题。 Esentially我希望能够在进入/离开协同程序的运行上下文中执行代码，在我的情况下，保持了各种各样的调用堆栈是正确的，即使在交错的情况下， yield秒。 事实证明，龙卷风支持StackContext的形式，可以使用如下形式：
@gen.coroutine
def correct():
yield run_with_stack_context(StackContext(ctx), other_coroutine)
其中ctx是在事件循环执行other_coroutine时将enter和exit的上下文管理器。
有关我如何使用它，请参阅https://github.com/muhrin/plumpy/blob/8d6cd97d8b521e42f124e77b08bb34c8375cd1b8/plumpy/processes.py#L467。
我没有看过实现，但龙卷风v5切换到使用asyncio作为它们的默认事件循环，所以它应该与此兼容。