asynchronous context manager for cancelling overdue coroutines use timeout or timeoutat rather than instantiating this class directly schedule a timeout that will trigger at a given loop time if when is none the timeout will never trigger if when loop time the timeout will trigger on the next iteration of the event loop return the current deadline return self when def rescheduleself when optionalfloat none is timeout expired during execution return self state in state expiring state expired def reprself str info if self state is state entered when roundself when 3 if self when is not none else none info appendfwhenwhen infostr joininfo return ftimeout self state valueinfostr async def aenterself timeout if self state is not state created raise runtimeerrortimeout has already been entered task tasks currenttask if task is none raise runtimeerrortimeout should be used inside a task self state state entered self task task self cancelling self task cancelling self rescheduleself when return self async def aexit self exctype optionaltypebaseexception excval optionalbaseexception exctb optionaltracebacktype optionalbool assert self state in state entered state expiring if self timeouthandler is not none self timeouthandler cancel self timeouthandler none if self state is state expiring self state state expired if self task uncancel self cancelling and exctype is exceptions cancellederror since there are no new cancel requests we re handling this raise timeouterror from excval elif self state is state entered self state state exited return none def ontimeoutself none assert self state is state entered self task cancel self state state expiring drop the reference early self timeouthandler none def timeoutdelay optionalfloat timeout loop events getrunningloop return timeoutloop time delay if delay is not none else none def timeoutatwhen optionalfloat timeout return timeoutwhen asynchronous context manager for cancelling overdue coroutines use timeout or timeout_at rather than instantiating this class directly schedule a timeout that will trigger at a given loop time if when is none the timeout will never trigger if when loop time the timeout will trigger on the next iteration of the event loop return the current deadline reschedule the timeout is timeout expired during execution since there are no new cancel requests we re handling this drop the reference early timeout async context manager useful in cases when you want to apply timeout logic around block of code or in cases when asyncio wait_for is not suitable for example async with asyncio timeout 10 10 seconds timeout await long_running_task delay value in seconds or none to disable timeout logic long_running_task is interrupted by raising asyncio cancellederror the top most affected timeout context manager converts cancellederror into timeouterror schedule the timeout at absolute time like timeout but argument gives absolute time in the same clock system as loop time please note it is not posix time but a time with undefined starting base e g the time of the system power on async with asyncio timeout_at loop time 10 await long_running_task when a deadline when timeout occurs or none to disable timeout logic long_running_task is interrupted by raising asyncio cancellederror the top most affected timeout context manager converts cancellederror into timeouterror
import enum

from types import TracebackType
from typing import final, Optional, Type

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"


@final
class Timeout:
    

    def __init__(self, when: Optional[float]) -> None:
        
        self._state = _State.CREATED

        self._timeout_handler: Optional[events.TimerHandle] = None
        self._task: Optional[tasks.Task] = None
        self._when = when

    def when(self) -> Optional[float]:
        
        return self._when

    def reschedule(self, when: Optional[float]) -> 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: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> Optional[bool]:
        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 exceptions.CancelledError:
                
                
                raise TimeoutError from exc_val
        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


def timeout(delay: Optional[float]) -> Timeout:
    
    loop = events.get_running_loop()
    return Timeout(loop.time() + delay if delay is not None else None)


def timeout_at(when: Optional[float]) -> Timeout:
    
    return Timeout(when)