向异步上下文管理器添加超时答案

【问题标题】：Adding a timeout to an Asynchronous Context Manager向异步上下文管理器添加超时
【发布时间】：2019-11-15 10:59:04
【问题描述】：

Python asynchronous context managers 很有用，但它们 do not work 和 asyncio.wait_for Timeouts。

将超时添加到异步上下文管理器的最佳方法是什么？

【问题讨论】：

【解决方案1】：

将超时添加到异步上下文管理器的最佳方法是什么？

您不能将wait_for 应用于异步上下文管理器，但可以将其应用于使用它的协程。因此，要将超时添加到上下文管理器，请在异步函数中使用它，然后对其应用超时。例如：

async def download(url, session):
    async with session.get(url) as resp:
        return await resp.text()

async def print_google(session):
    try:
        text = await asyncio.wait_for(download('http://www.google.com', session), 1)
    except asyncio.TimeoutError:
        text = None
    print(text)

【讨论】：

【解决方案2】：

你试过async-timeout吗？只需使用async with timeout() 包装任何异步代码（包括使用异步上下文管理器）：

import asyncio
from contextlib import asynccontextmanager

from async_timeout import timeout


@asynccontextmanager
async def my_acm():
    print('before')
    yield
    print('after')


async def main():
    async with timeout(1):
        async with my_acm():
            await asyncio.sleep(1.5)


asyncio.run(main())

如果您只想对异步上下文管理器应用超时，您可以创建利用 async-timeout 的新上下文管理器：

import asyncio
from contextlib import asynccontextmanager

from async_timeout import timeout


@asynccontextmanager
async def my_acm():
    print('before')
    yield
    print('after')


@asynccontextmanager
async def my_acm_plus_timeout(time):
    async with timeout(time):
        async with my_acm():
            yield


async def main():
    async with my_acm_plus_timeout(1):
        await asyncio.sleep(1.5)


asyncio.run(main())

【讨论】：