Python协程:暂停时释放上下文管理器 [英] Python coroutines: Release context manager when pausing
问题描述
背景:我是一位非常有经验的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
可以用协程吗?
推荐答案
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
这篇关于Python协程:暂停时释放上下文管理器的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!