Python `with` 上下文与生成器/协程/任务答案

【问题标题】：Python `with` context vs generators/coroutines/tasksPython `with` 上下文与生成器/协程/任务
【发布时间】：2017-04-18 00:23:58
【问题描述】：

我想尝试使用 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__（而不是像你一样不断地进入和退出）进出协程）。

有没有办法解决这个问题？是否有“coroutine-local”变量？

【问题讨论】：

没有全局可变状态（WithContext.stack 这里）；而是通过它。 for e in coroutine(stack): do_scoped_contextual_thing(stack) 其中stack 是不可变的。
你考虑过上下文结果的访问器吗？ with WithContext('outside') as outside: ... outside.do_scoped_contextual_thing() 之类的东西？
@Ryan 是的，我知道缺点。我还想尝试一下。
好吧，否则这是不可能的，所以这是替代方案......
我会注意到，@AustinHastings 的建议是 Python 已经如何处理这个问题。 decimal 模块确实允许 with 作用域上下文（基于线程本地，因此每个线程都有自己的上下文堆栈），但也使其大部分（全部？）API 可从上下文对象本身获得，因此您可以显式使用特定上下文而不替换线程的当前上下文。这里的解决方案是将您的 with 东西用于非协程使用，而协程使用显式引用其自身上下文的本地状态。

标签： python coroutine

【解决方案1】：

我对此感觉不太好，但我确实修改了您的测试代码以重新进入协程几次。与@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))

【讨论】：

【解决方案2】：

一个可能的半断“解决方案”是将上下文与堆栈帧的位置相关联，并在查找上下文时检查该位置。

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")

请注意，不断查找堆栈帧比仅仅传递值更昂贵，而且您可能不想告诉别人您写了这个。

请注意，这仍然会在更复杂的情况下中断。

例如：

如果同一方法在堆栈中出现两次怎么办？
如果生成器从一个地方迭代一点，然后从另一个地方再多一点迭代呢？
递归生成器呢？
异步方法呢？

【讨论】：

【解决方案3】：

我遇到了同样的问题。本质上，我希望能够在进入/离开协程的运行上下文时执行代码，在我的情况下，即使在交错的yields 的情况下也能保持正确的调用堆栈。事实证明tornado 以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 了解我的使用方法。

我没有研究实现，但 tornado v5 切换到使用 asyncio 作为默认事件循环，因此它也应该兼容。

【讨论】：