codekingpro/portable-devtools

1from __future__ import annotations2 3import math4from collections.abc import Generator5from contextlib import contextmanager6from types import TracebackType7 8from ..abc._tasks import TaskGroup, TaskStatus9from ._eventloop import get_async_backend10 11 12class _IgnoredTaskStatus(TaskStatus[object]):13    def started(self, value: object = None) -> None:14        pass15 16 17TASK_STATUS_IGNORED = _IgnoredTaskStatus()18 19 20class CancelScope:21    """22    Wraps a unit of work that can be made separately cancellable.23 24    :param deadline: The time (clock value) when this scope is cancelled automatically25    :param shield: ``True`` to shield the cancel scope from external cancellation26    :raises NoEventLoopError: if no supported asynchronous event loop is running in the27        current thread28    """29 30    def __new__(31        cls, *, deadline: float = math.inf, shield: bool = False32    ) -> CancelScope:33        return get_async_backend().create_cancel_scope(shield=shield, deadline=deadline)34 35    def cancel(self, reason: str | None = None) -> None:36        """37        Cancel this scope immediately.38 39        :param reason: a message describing the reason for the cancellation40 41        """42        raise NotImplementedError43 44    @property45    def deadline(self) -> float:46        """47        The time (clock value) when this scope is cancelled automatically.48 49        Will be ``float('inf')`` if no timeout has been set.50 51        """52        raise NotImplementedError53 54    @deadline.setter55    def deadline(self, value: float) -> None:56        raise NotImplementedError57 58    @property59    def cancel_called(self) -> bool:60        """``True`` if :meth:`cancel` has been called."""61        raise NotImplementedError62 63    @property64    def cancelled_caught(self) -> bool:65        """66        ``True`` if this scope suppressed a cancellation exception it itself raised.67 68        This is typically used to check if any work was interrupted, or to see if the69        scope was cancelled due to its deadline being reached. The value will, however,70        only be ``True`` if the cancellation was triggered by the scope itself (and not71        an outer scope).72 73        """74        raise NotImplementedError75 76    @property77    def shield(self) -> bool:78        """79        ``True`` if this scope is shielded from external cancellation.80 81        While a scope is shielded, it will not receive cancellations from outside.82 83        """84        raise NotImplementedError85 86    @shield.setter87    def shield(self, value: bool) -> None:88        raise NotImplementedError89 90    def __enter__(self) -> CancelScope:91        raise NotImplementedError92 93    def __exit__(94        self,95        exc_type: type[BaseException] | None,96        exc_val: BaseException | None,97        exc_tb: TracebackType | None,98    ) -> bool:99        raise NotImplementedError100 101 102@contextmanager103def fail_after(104    delay: float | None, shield: bool = False105) -> Generator[CancelScope, None, None]:106    """107    Create a context manager which raises a :class:`TimeoutError` if does not finish in108    time.109 110    :param delay: maximum allowed time (in seconds) before raising the exception, or111        ``None`` to disable the timeout112    :param shield: ``True`` to shield the cancel scope from external cancellation113    :return: a context manager that yields a cancel scope114    :rtype: :class:`~typing.ContextManager`\\[:class:`~anyio.CancelScope`\\]115    :raises NoEventLoopError: if no supported asynchronous event loop is running in the116        current thread117 118    """119    current_time = get_async_backend().current_time120    deadline = (current_time() + delay) if delay is not None else math.inf121    with get_async_backend().create_cancel_scope(122        deadline=deadline, shield=shield123    ) as cancel_scope:124        yield cancel_scope125 126    if cancel_scope.cancelled_caught and current_time() >= cancel_scope.deadline:127        raise TimeoutError128 129 130def move_on_after(delay: float | None, shield: bool = False) -> CancelScope:131    """132    Create a cancel scope with a deadline that expires after the given delay.133 134    :param delay: maximum allowed time (in seconds) before exiting the context block, or135        ``None`` to disable the timeout136    :param shield: ``True`` to shield the cancel scope from external cancellation137    :return: a cancel scope138    :raises NoEventLoopError: if no supported asynchronous event loop is running in the139        current thread140 141    """142    deadline = (143        (get_async_backend().current_time() + delay) if delay is not None else math.inf144    )145    return get_async_backend().create_cancel_scope(deadline=deadline, shield=shield)146 147 148def current_effective_deadline() -> float:149    """150    Return the nearest deadline among all the cancel scopes effective for the current151    task.152 153    :return: a clock value from the event loop's internal clock (or ``float('inf')`` if154        there is no deadline in effect, or ``float('-inf')`` if the current scope has155        been cancelled)156    :rtype: float157    :raises NoEventLoopError: if no supported asynchronous event loop is running in the158        current thread159 160    """161    return get_async_backend().current_effective_deadline()162 163 164def create_task_group() -> TaskGroup:165    """166    Create a task group.167 168    :return: a task group169    :raises NoEventLoopError: if no supported asynchronous event loop is running in the170        current thread171 172    """173    return get_async_backend().create_task_group()174