1  from __future__ import annotations
   2  
   3  import math
   4  from collections.abc import Generator
   5  from contextlib import contextmanager
   6  from types import TracebackType
   7  
   8  from ..abc._tasks import TaskGroup, TaskStatus
   9  from ._eventloop import get_async_backend
  10  
  11  
  12  class _IgnoredTaskStatus(TaskStatus[object]):
  13      def started(self, value: object = None) -> None:
  14          pass
  15  
  16  
  17  TASK_STATUS_IGNORED = _IgnoredTaskStatus()
  18  
  19  
  20  class 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 automatically
  25      :param shield: ``True`` to shield the cancel scope from external cancellation
  26      """
  27  
  28      def __new__(
  29          cls, *, deadline: float = math.inf, shield: bool = False
  30      ) -> CancelScope:
  31          return get_async_backend().create_cancel_scope(shield=shield, deadline=deadline)
  32  
  33      def cancel(self) -> None:
  34          """Cancel this scope immediately."""
  35          raise NotImplementedError
  36  
  37      @property
  38      def deadline(self) -> float:
  39          """
  40          The time (clock value) when this scope is cancelled automatically.
  41  
  42          Will be ``float('inf')`` if no timeout has been set.
  43  
  44          """
  45          raise NotImplementedError
  46  
  47      @deadline.setter
  48      def deadline(self, value: float) -> None:
  49          raise NotImplementedError
  50  
  51      @property
  52      def cancel_called(self) -> bool:
  53          """``True`` if :meth:`cancel` has been called."""
  54          raise NotImplementedError
  55  
  56      @property
  57      def cancelled_caught(self) -> bool:
  58          """
  59          ``True`` if this scope suppressed a cancellation exception it itself raised.
  60  
  61          This is typically used to check if any work was interrupted, or to see if the
  62          scope was cancelled due to its deadline being reached. The value will, however,
  63          only be ``True`` if the cancellation was triggered by the scope itself (and not
  64          an outer scope).
  65  
  66          """
  67          raise NotImplementedError
  68  
  69      @property
  70      def shield(self) -> bool:
  71          """
  72          ``True`` if this scope is shielded from external cancellation.
  73  
  74          While a scope is shielded, it will not receive cancellations from outside.
  75  
  76          """
  77          raise NotImplementedError
  78  
  79      @shield.setter
  80      def shield(self, value: bool) -> None:
  81          raise NotImplementedError
  82  
  83      def __enter__(self) -> CancelScope:
  84          raise NotImplementedError
  85  
  86      def __exit__(
  87          self,
  88          exc_type: type[BaseException] | None,
  89          exc_val: BaseException | None,
  90          exc_tb: TracebackType | None,
  91      ) -> bool:
  92          raise NotImplementedError
  93  
  94  
  95  @contextmanager
  96  def fail_after(
  97      delay: float | None, shield: bool = False
  98  ) -> Generator[CancelScope, None, None]:
  99      """
 100      Create a context manager which raises a :class:`TimeoutError` if does not finish in
 101      time.
 102  
 103      :param delay: maximum allowed time (in seconds) before raising the exception, or
 104          ``None`` to disable the timeout
 105      :param shield: ``True`` to shield the cancel scope from external cancellation
 106      :return: a context manager that yields a cancel scope
 107      :rtype: :class:`~typing.ContextManager`\\[:class:`~anyio.CancelScope`\\]
 108  
 109      """
 110      current_time = get_async_backend().current_time
 111      deadline = (current_time() + delay) if delay is not None else math.inf
 112      with get_async_backend().create_cancel_scope(
 113          deadline=deadline, shield=shield
 114      ) as cancel_scope:
 115          yield cancel_scope
 116  
 117      if cancel_scope.cancelled_caught and current_time() >= cancel_scope.deadline:
 118          raise TimeoutError
 119  
 120  
 121  def move_on_after(delay: float | None, shield: bool = False) -> CancelScope:
 122      """
 123      Create a cancel scope with a deadline that expires after the given delay.
 124  
 125      :param delay: maximum allowed time (in seconds) before exiting the context block, or
 126          ``None`` to disable the timeout
 127      :param shield: ``True`` to shield the cancel scope from external cancellation
 128      :return: a cancel scope
 129  
 130      """
 131      deadline = (
 132          (get_async_backend().current_time() + delay) if delay is not None else math.inf
 133      )
 134      return get_async_backend().create_cancel_scope(deadline=deadline, shield=shield)
 135  
 136  
 137  def current_effective_deadline() -> float:
 138      """
 139      Return the nearest deadline among all the cancel scopes effective for the current
 140      task.
 141  
 142      :return: a clock value from the event loop's internal clock (or ``float('inf')`` if
 143          there is no deadline in effect, or ``float('-inf')`` if the current scope has
 144          been cancelled)
 145      :rtype: float
 146  
 147      """
 148      return get_async_backend().current_effective_deadline()
 149  
 150  
 151  def create_task_group() -> TaskGroup:
 152      """
 153      Create a task group.
 154  
 155      :return: a task group
 156  
 157      """
 158      return get_async_backend().create_task_group()
 159