Сообщить об ошибке.

Асинхронный менеджер timeout() модуля asyncio в Python

Асинхронный диспетчер контекста для ограничения времени ожидания

Синтаксис:

import asyncio

# Новое в Python 3.11.
async with asyncio.timeout(delay):
    await long_running_task()

Параметры:

  • delay - число секунд ожидания. Может быть None или float/int.

Возвращаемое значение:

  • Асинхронный менеджер контекста asyncio.Timeout.

Описание:

Функция timeout() модуля asyncio представляет собой асинхронный диспетчер контекста, который можно использовать для ограничения времени, затрачиваемого на ожидание чего-либо.

Аргумент delay может быть None или числом секунд ожидания (float или int). Если задержка равна None, ограничение по времени применяться не будет; это может быть полезно, если задержка неизвестна при создании менеджера контекста.

Новое в Python 3.11.

Также обратите внимание на функцию asyncio.timeout_at(when) (новая в Python 3.11), которая создает асинхронный контекстный менеджер аналогично asyncio.timeout(), за исключением того, что аргумент when - это абсолютное время ожидания, или None.

В любом случае диспетчер контекста может быть перепланирован после создания с помощью метода Timeout.reschedule().

async def main():
    async with asyncio.timeout(10):
        await long_running_task()

Если для выполнения сопрограммы long_running_task() требуется более 10 секунд, то диспетчер контекста отменит текущую задачу и обработает полученную ошибку asyncio.CancelledError, преобразуя ее в ошибку asyncio.TimeoutError, которую можно перехватить и обработать.

Примечание. Контекстный менеджер asyncio.timeout() преобразует asyncio.CancelledError в asyncio.TimeoutError, что означает, что asyncio.TimeoutError может быть перехвачен только за пределами контекстного менеджера.

Пример отлова asyncio.TimeoutError:

async def main():
    try:
        async with asyncio.timeout(10):
            await long_running_task()
    except TimeoutError:
        print("Время длительной операции истекло.")

    print("print будет выполняться независимо.")

Менеджер контекста, созданный asyncio.timeout(), может быть перенесен на другой крайний срок и проверен.

Смотрите "Общий пример использования asyncio.timeout()".

Методы асинхронного менеджера контекста asyncio.Timeout.

Timeout.when():

Метод Timeout.when() возвращает текущий крайний срок или None, если текущий крайний срок не установлен.

Крайний срок - это число float, соответствующее времени, возвращаемому функцией loop.time().

Timeout.reschedule(when):

Метод Timeout.reschedule() изменяет время срабатывания тайм-аута.

  • Если when имеет значение None, то любой текущий крайний срок будет удален, и диспетчер контекста будет ждать неопределенное время.
  • Если это значение float, то оно устанавливается как новый крайний срок.

Если значение when уже прошло, то тайм-аут сработает на следующей итерации цикла событий.

Timeout.expired():

Метод Timeout.expired() возвращает True, если контекстный менеджер превысил свой крайний срок (т.е. время ожидания закончилось/истекло).

Общий пример использования asyncio.timeout().

async def main():
    try:
        # не знаем тайм-аута при запуске, поэтому передаем `None`.
        async with asyncio.timeout(None) as cm:
            # известен тайм-аут, поэтому переназначаем его.
            new_deadline = asyncio.get_running_loop().time() + 10
            cm.reschedule(new_deadline)

            await long_running_task()
    except TimeoutError:
        pass

    if cm.expired:
        print("Похоже, вовремя не закончили.")

Менеджеры контекста тайм-аута могут быть безопасно вложены друг в друга.

Новое в Python 3.11.