ULTRA [ growontech.com ]

Linux premium216.web-hosting.com 4.18.0-513.18.1.lve.el8.x86_64 #1 SMP Thu Feb 22 12:55:50 UTC 2024 x86_64

LiteSpeed

66.29.141.148 | Your IP: 18.117.8.177

Cant read [ /etc/named.conf ]

[ drwxr-xr-x ]

/ proc / thread-self / root / opt / alt / python312 / lib64 / python3.12 / asyncio /

Home Upload Mass Deface Mass Delete Back Connect Notify

timeouts.py

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:
    """Asynchronous context manager for cancelling overdue coroutines.

Use `timeout()` or `timeout_at()` rather than instantiating this class directly.
    """

def __init__(self, when: Optional[float]) -> None:
        """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.
        """
        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 the current deadline."""
        return self._when

def reschedule(self, when: Optional[float]) -> None:
        """Reschedule the timeout."""
        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:
        """Is timeout expired during execution?"""
        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:
                # 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) -> 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]) -> Timeout:
    """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.
    """
    loop = events.get_running_loop()
    return Timeout(loop.time() + delay if delay is not None else None)

def timeout_at(when: Optional[float]) -> Timeout:
    """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.
    """
    return Timeout(when)

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/linksafe	drwxr-xr-x	2024-10-23 05:05:12
..	dir	-	root/linksafe	drwxr-xr-x	2024-10-23 05:05:12
__pycache__	dir	-	root/linksafe	drwxr-xr-x	2024-10-23 05:05:12
__init__.py	text/plain	1.19 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
__main__.py	text/x-script.python	3.3 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
base_events.py	text/plain	76.1 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
base_futures.py	text/x-script.python	1.93 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
base_subprocess.py	text/x-script.python	8.66 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
base_tasks.py	text/x-script.python	2.61 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
constants.py	text/x-script.python	1.38 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
coroutines.py	text/x-script.python	3.26 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
events.py	text/plain	28.65 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
exceptions.py	text/plain	1.71 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
format_helpers.py	text/x-script.python	2.35 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
futures.py	text/plain	14.05 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
locks.py	text/plain	18.55 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
log.py	text/plain	124 B	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
mixins.py	text/plain	481 B	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
proactor_events.py	text/plain	32.71 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
protocols.py	text/plain	6.79 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
queues.py	text/x-script.python	7.79 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
runners.py	text/x-script.python	6.99 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
selector_events.py	text/plain	47.11 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
sslproto.py	text/x-script.python	31 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
staggered.py	text/plain	5.85 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
streams.py	text/x-script.python	26.97 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
subprocess.py	text/x-script.python	7.56 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
taskgroups.py	text/x-script.python	8.54 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
tasks.py	text/plain	36.49 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
threads.py	text/plain	790 B	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
timeouts.py	text/x-script.python	5.2 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
transports.py	text/plain	10.47 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
trsock.py	text/x-script.python	2.42 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
unix_events.py	text/plain	51.88 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
windows_events.py	text/plain	31.82 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53
windows_utils.py	text/plain	4.94 KB	root/linksafe	-rw-r--r--	2024-09-20 10:43:53