import asyncio # Новое в Python 3.11. async with asyncio.timeout(delay): await long_running_task()
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.