Как реализовать таймаут для выполнения корутины с помощью asyncio.wait_for, и что происходит с самой корутиной при его срабатывании?

Короткий ответ

asyncio.wait_for(coro, timeout=N) ожидает завершения корутины не дольше N секунд; если время истекло, внутренняя задача отменяется (cancel()), а сама функция поднимает TimeoutError, которую нужно явно обработать.

Как это работает подробнее

wait_for оборачивает переданную корутину в Task и запускает параллельно таймер; при срабатывании таймера вызывается cancel() у задачи, что инициирует CancelledError внутри корутины на следующей точке await.

Пример кода

example-273.pyКопировать

import asyncio
async def slow_external_call():
    await asyncio.sleep(5)
    return "response"
async def main():
    try:
        result = await asyncio.wait_for(slow_external_call(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("Таймаут: внешний сервис не ответил за 1 секунду")
asyncio.run(main())
# Проверка: вывод "Таймаут..." появляется примерно через 1с, а не через 5с

Что использовать на практике

• Практика. Таймауты критичны при обращении к внешним сервисам, чтобы не дать одному медленному вызову заблокировать обработку запроса на неопределённое время; ресурсы внутри отменённой корутины должны корректно освобождаться через try/finally.

Подводные камни

• Риски. Не обрабатывают TimeoutError, из-за чего исключение всплывает выше без понятного сообщения; забывают, что отменённая по таймауту корутина не «мгновенно умирает» — если внутри есть блокирующий код без await, отмена может не сработать сразу.