CODEWITHSUNDEEP
pythonpython-asyncio
Jethro
Jethro

Reputation: 3329

Adding a timeout to an Asynchronous Context Manager

Python asynchronous context managers are useful, but they do not work with asyncio.wait_for Timeouts.

What is the best way to add a Timeout to an asynchronous context manager?

Upvotes: 1

Views: 2362

Answers (2)

user4815162342
user4815162342

Reputation: 155046

What is the best way to add a Timeout to an asynchronous context manager?

You can't apply wait_for to an async context manager, but you can apply it to a coroutine that uses it. So to add timeout to a context manager, use it in an async function, and apply timeout to that. For example:

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)

Upvotes: 1

Mikhail Gerasimov
Mikhail Gerasimov

Reputation: 39546

Did you try async-timeout? Just wrap any async code (including usage of asynchronous context managers) with 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())

If you want to apply timeout to async context manager only, you can create new context manager that utilizes 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())

Upvotes: 1

Related Questions