codekingpro/portable-devtools

1from __future__ import annotations2 3from collections.abc import Awaitable, Generator4from typing import Any, cast5 6from ._eventloop import get_async_backend7 8 9class TaskInfo:10    """11    Represents an asynchronous task.12 13    :ivar int id: the unique identifier of the task14    :ivar parent_id: the identifier of the parent task, if any15    :vartype parent_id: Optional[int]16    :ivar str name: the description of the task (if any)17    :ivar ~collections.abc.Coroutine coro: the coroutine object of the task18    """19 20    __slots__ = "_name", "id", "parent_id", "name", "coro"21 22    def __init__(23        self,24        id: int,25        parent_id: int | None,26        name: str | None,27        coro: Generator[Any, Any, Any] | Awaitable[Any],28    ):29        func = get_current_task30        self._name = f"{func.__module__}.{func.__qualname__}"31        self.id: int = id32        self.parent_id: int | None = parent_id33        self.name: str | None = name34        self.coro: Generator[Any, Any, Any] | Awaitable[Any] = coro35 36    def __eq__(self, other: object) -> bool:37        if isinstance(other, TaskInfo):38            return self.id == other.id39 40        return NotImplemented41 42    def __hash__(self) -> int:43        return hash(self.id)44 45    def __repr__(self) -> str:46        return f"{self.__class__.__name__}(id={self.id!r}, name={self.name!r})"47 48    def has_pending_cancellation(self) -> bool:49        """50        Return ``True`` if the task has a cancellation pending, ``False`` otherwise.51 52        """53        return False54 55 56def get_current_task() -> TaskInfo:57    """58    Return the current task.59 60    :return: a representation of the current task61    :raises NoEventLoopError: if no supported asynchronous event loop is running in the62        current thread63 64    """65    return get_async_backend().get_current_task()66 67 68def get_running_tasks() -> list[TaskInfo]:69    """70    Return a list of running tasks in the current event loop.71 72    :return: a list of task info objects73    :raises NoEventLoopError: if no supported asynchronous event loop is running in the74        current thread75 76    """77    return cast("list[TaskInfo]", get_async_backend().get_running_tasks())78 79 80async def wait_all_tasks_blocked() -> None:81    """Wait until all other tasks are waiting for something."""82    await get_async_backend().wait_all_tasks_blocked()83