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

本文介绍了Python协程:暂停时释放上下文管理器的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

背景:我是一位非常有经验的Python程序员，他对新的协程/异步/等待功能一无所知.我无法编写一个异步的"hello world"来挽救我的生命.

Background: I'm a very experienced Python programmer who is completely clueless about the new coroutines/async/await features. I can't write an async "hello world" to save my life.

我的问题是:我得到了任意协程函数 f .我想编写一个协程函数 g 来包装 f ，即我将 g 给用户，就好像它是 f，并且用户会称它为明智"，因为 g 将在引擎盖下使用 f .就像装饰普通的Python函数以添加功能时一样.

My question is: I am given an arbitrary coroutine function f. I want to write a coroutine function g that will wrap f, i.e. I will give g to the user as if it was f, and the user will call it and be none the wiser, since g will be using f under the hood. Like when you decorate a normal Python function to add functionality.

我要添加的功能:每当程序流进入我的协程时，它都会获取我提供的上下文管理器，并且一旦程序流离开协程，它就会释放该上下文管理器.流量又回来了吗?重新获取上下文管理器.它退出了吗?重新释放它.直到协程完全完成.

The functionality that I want to add: Whenever the program flow goes into my coroutine, it acquires a context manager that I provide, and as soon as program flow goes out of the coroutine, it releases that context manager. Flow comes back in? Re-acquire the context manager. It goes back out? Re-release it. Until the coroutine is completely finished.

为了演示，这是使用普通生成器描述的功能:

To demonstrate, here is the described functionality with plain generators:

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

可以用协程吗?

推荐答案

协程是基于迭代器构建的- __ await __ 特殊方法是常规迭代器.这使您可以将基础迭代器包装在另一个迭代器中.诀窍是您必须使用目标的 __ await __ 来解包目标的迭代器，然后使用自己的来重新包装您自己的迭代器__await __ .

Coroutines are built on iterators - the __await__ special method is a regular iterator. This allows you to wrap the underlying iterator in yet another iterator. The trick is that you must unwrap the iterator of your target using its __await__, then re-wrap your own iterator using your own __await__.

适用于实例化协程的核心功能如下:

The core functionality that works on instantiated coroutines looks like this:

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.value 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 或 __ await __().throw .

Note that this explicitly works on a Coroutine, not an Awaitable - Coroutine.__await__ implements the generator interface. In theory, an Awaitable does not necessarily provide __await__().send or __await__().throw.

这足以传递进出消息:

import asyncio class PrintContext: def __enter__(self): print('enter') def __exit__(self, exc_type, exc_val, exc_tb): print('exit via', exc_type) return False async def main_coro(): print( 'wrapper returned', await CoroWrapper(test_coro(), PrintContext()) ) async def test_coro(delay=0.5): await asyncio.sleep(delay) return 2 asyncio.run(main_coro()) # enter # exit via None # enter # exit <class 'StopIteration'> # wrapper returned 2

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

---

You can delegate the wrapping part to a separate decorator. This also ensures that you have an actual coroutine, not a custom class - some async libraries require this.

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

这允许您直接装饰协程函数:

This allows you to directly decorate a coroutine function:

@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