File size: 2,217 Bytes
b6068b4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
from __future__ import annotations
from typing import Any, Awaitable, Generator
from ._compat import DeprecatedAwaitableList, _warn_deprecation
from ._eventloop import get_asynclib
class TaskInfo:
"""
Represents an asynchronous task.
:ivar int id: the unique identifier of the task
:ivar parent_id: the identifier of the parent task, if any
:vartype parent_id: Optional[int]
:ivar str name: the description of the task (if any)
:ivar ~collections.abc.Coroutine coro: the coroutine object of the task
"""
__slots__ = "_name", "id", "parent_id", "name", "coro"
def __init__(
self,
id: int,
parent_id: int | None,
name: str | None,
coro: Generator[Any, Any, Any] | Awaitable[Any],
):
func = get_current_task
self._name = f"{func.__module__}.{func.__qualname__}"
self.id: int = id
self.parent_id: int | None = parent_id
self.name: str | None = name
self.coro: Generator[Any, Any, Any] | Awaitable[Any] = coro
def __eq__(self, other: object) -> bool:
if isinstance(other, TaskInfo):
return self.id == other.id
return NotImplemented
def __hash__(self) -> int:
return hash(self.id)
def __repr__(self) -> str:
return f"{self.__class__.__name__}(id={self.id!r}, name={self.name!r})"
def __await__(self) -> Generator[None, None, TaskInfo]:
_warn_deprecation(self)
if False:
yield
return self
def _unwrap(self) -> TaskInfo:
return self
def get_current_task() -> TaskInfo:
"""
Return the current task.
:return: a representation of the current task
"""
return get_asynclib().get_current_task()
def get_running_tasks() -> DeprecatedAwaitableList[TaskInfo]:
"""
Return a list of running tasks in the current event loop.
:return: a list of task info objects
"""
tasks = get_asynclib().get_running_tasks()
return DeprecatedAwaitableList(tasks, func=get_running_tasks)
async def wait_all_tasks_blocked() -> None:
"""Wait until all other tasks are waiting for something."""
await get_asynclib().wait_all_tasks_blocked()
|