Python asyncio.timeout: ограничить время ожидания

asyncio.timeout() — контекстный менеджер из Python 3.11, который отменяет текущую задачу, если блок не завершился за указанное время. По выходу из блока выбрасывается встроенный TimeoutError. Это удобнее asyncio.wait_for(), когда нужно ограничить сразу несколько последовательных await, а не один вызов.

Прервать слишком долгий запрос

import asyncio

async def fetch_data(delay: float) -> str:
    await asyncio.sleep(delay)
    return "готово"

async def main() -> str:
    try:
        async with asyncio.timeout(0.5):
            return await fetch_data(1.0)
    except TimeoutError:
        return "превышен лимит ожидания"

print(asyncio.run(main()))  # превышен лимит ожидания

Таймаут отсчитывается от входа в блок и действует на все await внутри. При срабатывании отменяется задача, в которой выполняется блок, поэтому код после await может не выполниться. Очистку ресурсов делайте в finally или используйте async with для соединений и файлов.

Не отменять фоновую задачу

import asyncio

async def main() -> tuple[str, str]:
    task = asyncio.create_task(asyncio.sleep(2, result="задача завершилась"))
    try:
        async with asyncio.timeout(0.2):
            await asyncio.shield(task)
    except TimeoutError:
        note = "ожидание прервано"
    else:
        note = "успело"
    return note, await task

note, result = asyncio.run(main())
print(note)    # ожидание прервано
print(result)  # задача завершилась

asyncio.shield() не даёт отменить вложенную задачу, но внешний await всё равно завершается с TimeoutError. После обработки исключения задачу можно дождаться и забрать результат. Если не сохранить ссылку на задачу, её может собрать сборщик мусора.

Не ловите asyncio.CancelledError как обычную ошибку: это сигнал отмены, и его нужно либо пробросить, либо корректно завершить работу. asyncio.timeout() не прерывает синхронный код и блокирующие вызовы — они не отдают управление event loop. На Python 3.10 и раньше используйте asyncio.wait_for() с явным таймаутом.

Документация: asyncio.timeout, asyncio.shield.