Функция asyncio.wait_for() в Python с примерами

В этом руководстве вы узнаете, как использовать функцию asyncio.wait_for() для ожидания завершения сопрограммы с таймаутом в Python.

Что такое функция asyncio.wait_for() в Python?

Ранее вы узнали, как отменить выполняемую задачу с помощью метода cancel() объекта Task.

Чтобы дождаться завершения задачи с таймаутом, вы можете использовать функцию asyncio.wait_for(). Функция asyncio.wait_for() ожидает завершения одной задачи с таймаутом.

При возникновении таймаута функция asyncio.wait_for() отменяет задачу и вызывает исключение TimeoutError. В противном случае она возвращает результат задачи. Например:

import asyncio
from asyncio.exceptions import TimeoutError


async def call_api(message, result=1000, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result


async def main():
    task = asyncio.create_task(
        call_api('Calling API...', result=2000, delay=5)
    )

    MAX_TIMEOUT = 3
    try:
        await asyncio.wait_for(task, timeout=MAX_TIMEOUT)
    except TimeoutError:
        print('The task was cancelled due to a timeout')

asyncio.run(main())

Выход:

Calling API...
The task was cancelled due to a timeout

Как это работает

• Сначала определите сопрограмму call_api(), выполнение которой по умолчанию занимает 3 секунды:

async def call_api(message, result=1000, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result

• Во-вторых, создайте задачу, которая оборачивает сопрограмму call_api и занимает 5 секунд:

task = asyncio.create_task(
    call_api('Calling API...', result=2000, delay=5)
)

• В-третьих, используйте функцию asyncio.wait_for(), чтобы дождаться завершения задачи с таймаутом в 3 секунды. Поскольку выполнение задачи занимает 5 секунд, произойдет таймаут и возникнет ошибка TimeoutError:

MAX_TIMEOUT = 3
try:
    await asyncio.wait_for(task, timeout=MAX_TIMEOUT)
except TimeoutError:
    print('The task was cancelled due to a timeout')

Защита задачи от отмены

Иногда вам может потребоваться сообщить пользователям, что задача выполняется дольше, чем ожидалось, по истечении определенного периода времени, но не отменять задачу при превышении времени ожидания.

Для этого вы можете применить функцию asyncio.shield(). Она предотвращает отмену задачи. Например:

import asyncio
from asyncio.exceptions import TimeoutError


async def call_api(message, result=1000, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result


async def main():
    task = asyncio.create_task(
        call_api('Calling API...', result=2000, delay=5)
    )

    MAX_TIMEOUT = 3
    try:
        await asyncio.wait_for(asyncio.shield(task), timeout=MAX_TIMEOUT)
    except TimeoutError:
        print('The task took more than expected and will complete soon.')
        result = await task
        print(result)

asyncio.run(main())

Выход:

Calling API...
The task took more than expected and will complete soon.
2000

В этом примере выполнение задачи занимает 5 секунд. Когда таймаут достигает 3 секунд, возникает исключение TimeoutEror. Однако задача не отменяется благодаря функции asyncio.shield().

В разделе обработки исключений мы ожидаем завершения задачи и распечатываем результат.