python-asyncio - 如何在 Python 3 中嵌套异步上下文管理器

Question

假设我们有两个异步上下文管理器，它们通常以嵌套方式一起使用，但通常只有第二个的结果在正文中使用。例如，如果我们发现自己经常输入以下内容:

async with context_mgr_1() as cm1:
    async with cm2.context_mgr_2() as cm2:
        ...do something with cm2...

我们如何创建一个嵌套这些上下文管理器的单个上下文管理器，以便我们可以这样做:

async with context_mgr_2() as cm2:
    ...do something with cm2...

contextlib.nested 用于为非异步上下文管理器完成此操作，但我在 asyncio 中找不到这样的帮助程序。

Answer 1

由于 Python 3.7 有 contextlib.AsyncExitStack :

async with AsyncExitStack() as stack:
    cm1 = await stack.enter_async_context(context_mgr_1())
    cm2 = await stack.enter_async_context(context_mgr_2())
    # ...do something with cm2...

它可以动态使用:

async with AsyncExitStack() as stack:
    connections = [await stack.enter_async_context(get_connection())
        for i in range(5)]
    # All opened connections will automatically be released at the end of
    # the async with statement, even if attempts to open a connection
    # later in the list raise an exception.

它继承了 contextlib.ExitStack所以你可以结合异步和非异步上下文管理器:

async with AsyncExitStack() as stack:
    cm1 = await stack.enter_async_context(context_mgr_1())
    f = stack.enter_context(open('file.txt'))
    cm2 = await stack.enter_async_context(context_mgr_2())
    # ...