Python协程：暂停时释放上下文管理器

Question

背景：我是一个经验丰富的Python程序员，对新的coroutines/async/await特性一无所知。我不能写一个异步的“hello world”来救我的命。在

我的问题是：我得到了一个任意的协同程序函数f。我想写一个协同程序函数g，它将包装f，也就是说，我将把g交给用户，就像它是{}一样，用户会调用它，而且不会更聪明，因为g将在幕后使用{}。就像你装饰一个普通的Python函数来添加功能一样。在

我想添加的功能：每当程序流进入我的协同程序时，它就会获得我提供的上下文管理器，一旦程序流离开协同程序，它就会释放上下文管理器。流回来了？重新获取上下文管理器。它又出来了？重新释放它。直到合作计划完成。在

为了演示，下面介绍了普通生成器的功能：

def generator_wrapper(_, *args, **kwargs):
    gen = function(*args, **kwargs)
    method, incoming = gen.send, None
    while True:
        with self:
            outgoing = method(incoming)
        try:
            method, incoming = gen.send, (yield outgoing)
        except Exception as e:
            method, incoming = gen.throw, e

可以用协同程序来完成吗？

Answer 1

协程建立在迭代器上，^{} special method是一个常规迭代器。这允许您将底层迭代器包装在另一个迭代器中。诀窍在于，必须使用目标的迭代器__await__展开，然后使用自己的__await__重新包装自己的迭代器。在

用于实例化协同程序的核心功能如下所示：

class CoroWrapper:
    """Wrap ``target`` to have every send issued in a ``context``"""
    def __init__(self, target: 'Coroutine', context: 'ContextManager'):
        self.target = target
        self.context = context

    # wrap an iterator for use with 'await'
    def __await__(self):
        # unwrap the underlying iterator
        target_iter = self.target.__await__()
        # emulate 'yield from'
        iter_send, iter_throw = target_iter.send, target_iter.throw
        send, message = iter_send, None
        while True:
            # communicate with the target coroutine
            try:
                with self.context:
                    signal = send(message)
            except StopIteration as err:
                return err.args[0] if err.args else None
            else:
                send = iter_send
            # communicate with the ambient event loop
            try:
                message = yield signal
            except BaseException as err:
                send, message = iter_throw, err

注意，这在Coroutine上显式工作，而不是Awaitable-Coroutine.__await__实现生成器接口。理论上，Awaitable不一定提供__await__().send或{}。在

这足以将消息传入和传出：

^{pr2}$

---

您可以将包装部分委托给一个单独的装饰器。这还可以确保您有一个实际的协同程序，而不是一个自定义类-一些异步库需要这样做。在

from functools import wraps


def send_context(context: 'ContextManager'):
    """Wrap a coroutine to issue every send in a context"""
    def coro_wrapper(target: 'Callable[..., Coroutine]') -> 'Callable[..., Coroutine]':
        @wraps(target)
        async def context_coroutine(*args, **kwargs):
            return await CoroWrapper(target(*args, **kwargs), context)
        return context_coroutine
    return coro_wrapper

这允许您直接装饰协同程序函数：

@send_context(PrintContext())
async def test_coro(delay=0.5):
    await asyncio.sleep(delay)
    return 2

print('async run returned:', asyncio.run(test_coro()))
# enter
# exit via None
# enter
# exit via <class 'StopIteration'>
# async run returned: 2