python - 取消异步上下文管理器

Question

获取要取消的任务的句柄非常容易:

task = loop.create_task(coro_fn())

# later
task.cancel()

是否可以对异步上下文管理器执行相同的操作？

async with foo() as bar:
   # is it possible to cancel before getting here,
   # while waiting to enter the context manager?
   await bar.baz()

# later
# How do I get a handle to the context manager for cancellation?

有什么办法可以做到这一点吗？或者上下文管理器代码是否需要在自己的任务中运行？

Answer 1

How do I get a handle to the context manager for cancellation?

你不知道，至少不直接知道 - 上下文管理器只是获取和释放资源的方法的方便包装。您的问题中的 async with foo() as bar: ... 大致如下:

_cm = foo()
bar = await _cm.__aenter__()
try:
    await bar.baz()
finally:
    await _cm.__aexit__(*sys.exc_info())

__aenter__和__aexit__都是普通的awaitable，其执行可以被取消，就像await bar.baz的执行一样() 在 async with 主体内。

因此，如果“取消上下文管理器”是指取消 async with block 当前陷入的操作，则可以像任何其他取消一样执行此操作。例如，您可以将async with提取到在其自己的任务中运行的协程中，然后cancel()它。请注意，无论如何您都需要获取一个任务才能到达“稍后”部分，而异步上下文管理器在此过程中不会关闭。

这是一个例子:

async def run_foo():
    async with foo() as bar:
       # is it possible to cancel before getting here,
       # while waiting to enter the context manager?
       await bar.baz()

task = asyncio.create_task(run_foo())
# "task" is the handle you can await or cancel at your leisure