<h2>Editing File: /lib64/python3.12/asyncio/timeouts.py</h2><form method='post'><textarea name='content' class='file-edit'>import enum

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

from . import events
from . import exceptions
from . import tasks


__all__ = (
    &quot;Timeout&quot;,
    &quot;timeout&quot;,
    &quot;timeout_at&quot;,
)


class _State(enum.Enum):
    CREATED = &quot;created&quot;
    ENTERED = &quot;active&quot;
    EXPIRING = &quot;expiring&quot;
    EXPIRED = &quot;expired&quot;
    EXITED = &quot;finished&quot;


@final
class Timeout:
    &quot;&quot;&quot;Asynchronous context manager for cancelling overdue coroutines.

    Use `timeout()` or `timeout_at()` rather than instantiating this class directly.
    &quot;&quot;&quot;

    def __init__(self, when: Optional[float]) -&gt; None:
        &quot;&quot;&quot;Schedule a timeout that will trigger at a given loop time.

        - If `when` is `None`, the timeout will never trigger.
        - If `when &lt; loop.time()`, the timeout will trigger on the next
          iteration of the event loop.
        &quot;&quot;&quot;
        self._state = _State.CREATED

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

    def when(self) -&gt; Optional[float]:
        &quot;&quot;&quot;Return the current deadline.&quot;&quot;&quot;
        return self._when

    def reschedule(self, when: Optional[float]) -&gt; None:
        &quot;&quot;&quot;Reschedule the timeout.&quot;&quot;&quot;
        if self._state is not _State.ENTERED:
            if self._state is _State.CREATED:
                raise RuntimeError(&quot;Timeout has not been entered&quot;)
            raise RuntimeError(
                f&quot;Cannot change state of {self._state.value} Timeout&quot;,
            )

        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 &lt;= 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) -&gt; bool:
        &quot;&quot;&quot;Is timeout expired during execution?&quot;&quot;&quot;
        return self._state in (_State.EXPIRING, _State.EXPIRED)

    def __repr__(self) -&gt; str:
        info = ['']
        if self._state is _State.ENTERED:
            when = round(self._when, 3) if self._when is not None else None
            info.append(f&quot;when={when}&quot;)
        info_str = ' '.join(info)
        return f&quot;&lt;Timeout [{self._state.value}]{info_str}&gt;&quot;

    async def __aenter__(self) -&gt; &quot;Timeout&quot;:
        if self._state is not _State.CREATED:
            raise RuntimeError(&quot;Timeout has already been entered&quot;)
        task = tasks.current_task()
        if task is None:
            raise RuntimeError(&quot;Timeout should be used inside a task&quot;)
        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],
    ) -&gt; 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() &lt;= self._cancelling and exc_type is exceptions.CancelledError:
                # Since there are no new cancel requests, we're
                # handling this.
                raise TimeoutError from exc_val
        elif self._state is _State.ENTERED:
            self._state = _State.EXITED

        return None

    def _on_timeout(self) -&gt; None:
        assert self._state is _State.ENTERED
        self._task.cancel()
        self._state = _State.EXPIRING
        # drop the reference early
        self._timeout_handler = None


def timeout(delay: Optional[float]) -&gt; Timeout:
    &quot;&quot;&quot;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:

    &gt;&gt;&gt; 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.
    &quot;&quot;&quot;
    loop = events.get_running_loop()
    return Timeout(loop.time() + delay if delay is not None else None)


def timeout_at(when: Optional[float]) -&gt; Timeout:
    &quot;&quot;&quot;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.

    &gt;&gt;&gt; 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.
    &quot;&quot;&quot;
    return Timeout(when)
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Flib64%2Fpython3.12%2Fasyncio'>Back to Directory</a>