timeouts.py
python
import enum
from types import TracebackType
from . import events
from . import exceptions
from . import tasks
__all__ = (
"Timeout",
"timeout",
"timeout_at",
)
class _State(enum.Enum):
CREATED = "created"
ENTERED = "active"
EXPIRING = "expiring"
EXPIRED = "expired"
EXITED = "finished"
class Timeout:
"""Отправить данные на транспорт.
Этот метод не блокируется; он буферизует данные и организует их
асинхронную отправку.
addr — адрес целевого сокета.
Если addr равен None, используется целевой адрес, указанный при создании транспорта.
Если data является пустым объектом bytes, будет отправлена датаграмма нулевой длины.
"""
def __init__(self, when: float | None) -> None:
"""Запланировать тайм-аут, который сработает в заданное время цикла.
- Если `when` равно `None`, тайм-аут никогда не сработает.
- Если `when < loop.time()`, тайм-аут сработает на следующей
итерации цикла событий.
"""
self._state = _State.CREATED
self._timeout_handler: events.TimerHandle | None = None
self._task: tasks.Task | None = None
self._when = when
def when(self) -> float | None:
"""Возвращает текущий крайний срок."""
return self._when
def reschedule(self, when: float | None) -> None:
"""Переназначить тайм-аут."""
if self._state is not _State.ENTERED:
if self._state is _State.CREATED:
raise RuntimeError("Timeout has not been entered")
raise RuntimeError(
f"Cannot change state of {self._state.value} Timeout",
)
self._when = when
if self._timeout_handler is not None:
self._timeout_handler.cancel()
if when is None:
self._timeout_handler = None
else:
loop = events.get_running_loop()
if when <= loop.time():
self._timeout_handler = loop.call_soon(self._on_timeout)
else:
self._timeout_handler = loop.call_at(when, self._on_timeout)
def expired(self) -> bool:
"""Истёк ли тайм-аут во время выполнения?"""
return self._state in (_State.EXPIRING, _State.EXPIRED)
def __repr__(self) -> str:
info = ['']
if self._state is _State.ENTERED:
when = round(self._when, 3) if self._when is not None else None
info.append(f"when={when}")
info_str = ' '.join(info)
return f"<Timeout [{self._state.value}]{info_str}>"
async def __aenter__(self) -> "Timeout":
if self._state is not _State.CREATED:
raise RuntimeError("Timeout has already been entered")
task = tasks.current_task()
if task is None:
raise RuntimeError("Timeout should be used inside a task")
self._state = _State.ENTERED
self._task = task
self._cancelling = self._task.cancelling()
self.reschedule(self._when)
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> bool | None:
assert self._state in (_State.ENTERED, _State.EXPIRING)
if self._timeout_handler is not None:
self._timeout_handler.cancel()
self._timeout_handler = None
if self._state is _State.EXPIRING:
self._state = _State.EXPIRED
if self._task.uncancel() <= self._cancelling and exc_type is not None:
# Поскольку нет новых запросов на отмену, мы
# обрабатываем это.
if issubclass(exc_type, exceptions.CancelledError):
raise TimeoutError from exc_val
elif exc_val is not None:
self._insert_timeout_error(exc_val)
if isinstance(exc_val, ExceptionGroup):
for exc in exc_val.exceptions:
self._insert_timeout_error(exc)
elif self._state is _State.ENTERED:
self._state = _State.EXITED
return None
def _on_timeout(self) -> None:
assert self._state is _State.ENTERED
self._task.cancel()
self._state = _State.EXPIRING
# освобождаем ссылку раньше
self._timeout_handler = None
@staticmethod
def _insert_timeout_error(exc_val: BaseException) -> None:
while exc_val.__context__ is not None:
if isinstance(exc_val.__context__, exceptions.CancelledError):
te = TimeoutError()
te.__context__ = te.__cause__ = exc_val.__context__
exc_val.__context__ = te
break
exc_val = exc_val.__context__
def timeout(delay: float | None) -> Timeout:
"""Тайм-аут асинхронного контекстного менеджера.
Полезен в случаях, когда нужно применить логику тайм-аута к блоку кода или когда asyncio.wait_for не подходит. Например:
>>> async with asyncio.timeout(10): # 10 seconds timeout
... await long_running_task()
delay - значение в секундах или None для отключения логики тайм-аута
long_running_task() прерывается вызовом asyncio.CancelledError, самый верхний затронутый контекстный менеджер timeout() преобразует CancelledError в TimeoutError.
"""
loop = events.get_running_loop()
return Timeout(loop.time() + delay if delay is not None else None)
def timeout_at(when: float | None) -> Timeout:
"""Устанавливает тайм-аут на абсолютное время.
Подобно timeout(), но аргумент задает абсолютное время в той же системе часов, что и loop.time().
Обратите внимание: это не время POSIX, а время с неопределенной начальной точкой отсчета, например, время с момента включения системы.
>>> async with asyncio.timeout_at(loop.time() + 10):
... await long_running_task()
when - крайний срок, когда наступает тайм-аут, или None для отключения логики тайм-аута
long_running_task() прерывается вызовом asyncio.CancelledError, самый верхний затронутый контекстный менеджер timeout() преобразует CancelledError в TimeoutError.
"""
return Timeout(when)