← back to Atmos Game

assets/gen/.venv/lib/python3.12/site-packages/anyio/_core/_tasks.py

413 lines

from __future__ import annotations

import math
import sys
from collections.abc import (
    Coroutine,
    Generator,
)
from contextlib import (
    contextmanager,
)
from enum import Enum, auto
from inspect import iscoroutine
from types import TracebackType
from typing import Any, Generic, final

from ..abc import TaskGroup, TaskStatus
from ._eventloop import get_async_backend, get_cancelled_exc_class
from ._exceptions import TaskCancelled, TaskFailed, TaskNotFinished

if sys.version_info >= (3, 13):
    from typing import TypeVar
else:
    from typing_extensions import TypeVar

if sys.version_info >= (3, 11):
    from typing import Never, TypeVarTuple
else:
    from typing_extensions import Never, TypeVarTuple

T = TypeVar("T")
T_co = TypeVar("T_co", covariant=True)
T_startval = TypeVar("T_startval", covariant=True, default=Never)
PosArgsT = TypeVarTuple("PosArgsT")


class _IgnoredTaskStatus(TaskStatus[object]):
    def started(self, value: object = None) -> None:
        pass


TASK_STATUS_IGNORED = _IgnoredTaskStatus()


class CancelScope:
    """
    Wraps a unit of work that can be made separately cancellable.

    :param deadline: The time (clock value) when this scope is cancelled automatically
    :param shield: ``True`` to shield the cancel scope from external cancellation
    :raises NoEventLoopError: if no supported asynchronous event loop is running in the
        current thread
    """

    __slots__ = ("__weakref__",)

    def __new__(
        cls, *, deadline: float = math.inf, shield: bool = False
    ) -> CancelScope:
        return get_async_backend().create_cancel_scope(shield=shield, deadline=deadline)

    def cancel(self, reason: str | None = None) -> None:
        """
        Cancel this scope immediately.

        :param reason: a message describing the reason for the cancellation

        """
        raise NotImplementedError

    @property
    def deadline(self) -> float:
        """
        The time (clock value) when this scope is cancelled automatically.

        Will be ``float('inf')`` if no timeout has been set.

        """
        raise NotImplementedError

    @deadline.setter
    def deadline(self, value: float) -> None:
        raise NotImplementedError

    @property
    def cancel_called(self) -> bool:
        """``True`` if :meth:`cancel` has been called."""
        raise NotImplementedError

    @property
    def cancelled_caught(self) -> bool:
        """
        ``True`` if this scope suppressed a cancellation exception it itself raised.

        This is typically used to check if any work was interrupted, or to see if the
        scope was cancelled due to its deadline being reached. The value will, however,
        only be ``True`` if the cancellation was triggered by the scope itself (and not
        an outer scope).

        """
        raise NotImplementedError

    @property
    def shield(self) -> bool:
        """
        ``True`` if this scope is shielded from external cancellation.

        While a scope is shielded, it will not receive cancellations from outside.

        """
        raise NotImplementedError

    @shield.setter
    def shield(self, value: bool) -> None:
        raise NotImplementedError

    def __enter__(self) -> CancelScope:
        raise NotImplementedError

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> bool:
        raise NotImplementedError


@contextmanager
def fail_after(
    delay: float | None, shield: bool = False
) -> Generator[CancelScope, None, None]:
    """
    Create a context manager which raises a :class:`TimeoutError` if does not finish in
    time.

    :param delay: maximum allowed time (in seconds) before raising the exception, or
        ``None`` to disable the timeout
    :param shield: ``True`` to shield the cancel scope from external cancellation
    :return: a context manager that yields a cancel scope
    :rtype: :class:`~typing.ContextManager`\\[:class:`~anyio.CancelScope`\\]
    :raises NoEventLoopError: if no supported asynchronous event loop is running in the
        current thread

    """
    current_time = get_async_backend().current_time
    deadline = (current_time() + delay) if delay is not None else math.inf
    with get_async_backend().create_cancel_scope(
        deadline=deadline, shield=shield
    ) as cancel_scope:
        yield cancel_scope

    if cancel_scope.cancelled_caught and current_time() >= cancel_scope.deadline:
        raise TimeoutError


def move_on_after(delay: float | None, shield: bool = False) -> CancelScope:
    """
    Create a cancel scope with a deadline that expires after the given delay.

    :param delay: maximum allowed time (in seconds) before exiting the context block, or
        ``None`` to disable the timeout
    :param shield: ``True`` to shield the cancel scope from external cancellation
    :return: a cancel scope
    :raises NoEventLoopError: if no supported asynchronous event loop is running in the
        current thread

    """
    deadline = (
        (get_async_backend().current_time() + delay) if delay is not None else math.inf
    )
    return get_async_backend().create_cancel_scope(deadline=deadline, shield=shield)


def current_effective_deadline() -> float:
    """
    Return the nearest deadline among all the cancel scopes effective for the current
    task.

    :return: a clock value from the event loop's internal clock (or ``float('inf')`` if
        there is no deadline in effect, or ``float('-inf')`` if the current scope has
        been cancelled)
    :rtype: float
    :raises NoEventLoopError: if no supported asynchronous event loop is running in the
        current thread

    """
    return get_async_backend().current_effective_deadline()


def create_task_group() -> TaskGroup:
    """
    Create a task group.

    :return: a task group
    :raises NoEventLoopError: if no supported asynchronous event loop is running in the
        current thread

    """
    return get_async_backend().create_task_group()


@final
class TaskHandle(Generic[T_co, T_startval]):
    """
    Returned from the task-spawning methods of :class:`TaskGroup`. Can be awaited on to
    get the return value of the task (or the raised exception). If the task was
    terminated by a :exc:`BaseException`, :exc:`TaskFailed` will be raised (or its
    subclass :exc:`TaskCancelled` if the task was cancelled).

    .. versionadded:: 4.14.0
    """

    class Status(Enum):
        """
        The status of a task handle.

        .. attribute:: PENDING

            The task has not finished yet.
        .. attribute:: FINISHED

            The task has finished with a return value.
        .. attribute:: CANCELLING

            The task has been cancelled but has not finished yet.
        .. attribute:: CANCELLED

            The task was cancelled and has finished since.
        .. attribute:: FAILED

            The task raised an exception.
        """

        PENDING = auto()
        FINISHED = auto()
        CANCELLING = auto()
        CANCELLED = auto()
        FAILED = auto()

    __slots__ = (
        "__weakref__",
        "_coro",
        "_name",
        "_cancel_scope",
        "_finished_event",
        "_return_value",
        "_start_value",
        "_exception",
    )

    _return_value: T_co
    _start_value: T_startval

    def __init__(self, coro: Coroutine[Any, Any, T_co], name: object) -> None:
        from ._synchronization import Event

        self._coro = coro
        self._cancel_scope = CancelScope()
        self._finished_event = Event()
        self._exception: BaseException | None = None

        if name is not None:
            self._name = str(name)
        elif iscoroutine(coro):
            self._name = coro.__qualname__
        else:
            self._name = str(coro)  # coroutine-like object (e.g. asend() objects)

    async def _run_coro(self) -> None:
        __tracebackhide__ = True

        with self._cancel_scope:
            try:
                retval = await self._coro
            except BaseException as exc:
                self._exception = exc
                raise
            else:
                self._return_value = retval
            finally:
                self._finished_event.set()
                del self  # Break the reference cycle

    def cancel(self) -> None:
        """
        Set the task to a cancelled state.

        This will interrupt any interruptible asynchronous operation, and will cause
        any further awaits on this task to get immediately cancelled, unless done in
        a shielded cancel scope.

        If the task has already finished, this method has no effect.
        """
        if not self._finished_event.is_set():
            self._cancel_scope.cancel()

    @property
    def coro(self) -> Coroutine[Any, Any, T_co]:
        """
        The coroutine object that was passed to one of the task-spawning methods in
        :class:`TaskGroup`.
        """
        return self._coro

    @property
    def status(self) -> TaskHandle.Status:
        """
        The current status of the task.

        Every task starts in the :attr:`~TaskHandle.Status.PENDING` state.
        If a task is cancelled while in this state, it will transition to the
        :attr:`~TaskHandle.Status.CANCELLING` state. When the task finishes, it will
        transition to one of the three final states (
        :attr:`~TaskHandle.Status.FINISHED`, :attr:`~TaskHandle.Status.FAILED`, or
        :attr:`~TaskHandle.Status.CANCELLING`) depending on the exception the task
        raised, if any. No other status transitions will happen.
        """
        if not self._finished_event.is_set():
            if self._cancel_scope.cancel_called:
                return TaskHandle.Status.CANCELLING
            else:
                return TaskHandle.Status.PENDING
        elif self._exception is not None:
            if isinstance(self._exception, get_cancelled_exc_class()):
                return TaskHandle.Status.CANCELLED
            else:
                return TaskHandle.Status.FAILED
        else:
            return TaskHandle.Status.FINISHED

    @property
    def name(self) -> str:
        """The name of the task."""
        return self._name

    @property
    def exception(self) -> BaseException | None:
        """
        The exception raised by the task, or ``None`` if it finished without raising.

        :raises TaskNotFinished: if the task has not finished yet
        :raises TaskCancelled: if the task was cancelled

        """
        match self.status:
            case TaskHandle.Status.PENDING:
                raise TaskNotFinished("the task has not finished yet")
            case TaskHandle.Status.FINISHED:
                return None
            case TaskHandle.Status.CANCELLING:
                raise TaskCancelled("the task was cancelled")
            case TaskHandle.Status.CANCELLED:
                raise TaskCancelled("the task was cancelled") from self._exception
            case TaskHandle.Status.FAILED:
                return self._exception

    @property
    def return_value(self) -> T_co:
        """
        The return value of the task.

        :raises TaskNotFinished: if the task has not finished yet
        :raises TaskCancelled: if the task was cancelled
        :raises TaskFailed: if the task raised an exception

        """
        match self.status:
            case TaskHandle.Status.PENDING:
                raise TaskNotFinished("the task has not finished yet")
            case TaskHandle.Status.FINISHED:
                return self._return_value
            case TaskHandle.Status.CANCELLING:
                raise TaskCancelled("the task was cancelled")
            case TaskHandle.Status.CANCELLED:
                raise TaskCancelled("the task was cancelled") from self._exception
            case TaskHandle.Status.FAILED:
                raise TaskFailed("the task raised an exception") from self._exception

    @property
    def start_value(self) -> T_startval:
        """
        The value passed to :meth:`task_status.started() <.abc.TaskStatus.started>`,

        :raises RuntimeError: if the task was not started with :meth:`TaskGroup.start()
            <.abc.TaskGroup.start>`
        """
        try:
            return self._start_value
        except AttributeError:
            raise RuntimeError(
                "the task was not started with TaskGroup.start()"
            ) from None

    async def wait(self) -> None:
        """
        Wait for the task to finish.

        This method will return as soon as the task has finished, no matter how it
        happened.
        """
        await self._finished_event.wait()

    def __await__(self) -> Generator[Any, Any, T_co]:
        yield from self._finished_event.wait().__await__()
        return self.return_value

    def __repr__(self) -> str:
        return (
            f"<{self.__class__.__name__} {self.status.name.lower()} "
            f"name={self._name!r} coro={self._coro!r}>"
        )