| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209 |
- from __future__ import annotations
- import sys
- from abc import ABCMeta, abstractmethod
- from collections.abc import Callable, Coroutine
- from contextvars import Context
- from types import TracebackType
- from typing import TYPE_CHECKING, Any, Literal, Protocol, final, overload
- 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 TypeVarTuple, Unpack
- else:
- from typing_extensions import TypeVarTuple, Unpack
- if TYPE_CHECKING:
- from .._core._tasks import CancelScope, TaskHandle
- T_co = TypeVar("T_co", covariant=True)
- T_contra = TypeVar("T_contra", contravariant=True, default=None)
- PosArgsT = TypeVarTuple("PosArgsT")
- def get_callable_name(func: Callable, override: object = None) -> str:
- if override is not None:
- return str(override)
- module = getattr(func, "__module__", None)
- qualname = getattr(func, "__qualname__", None)
- return ".".join([x for x in (module, qualname) if x])
- def call_for_coroutine(
- func: Callable[[Unpack[PosArgsT]], Coroutine[Any, Any, T_co]],
- args: tuple[Unpack[PosArgsT]],
- **kwargs: Any,
- ) -> Coroutine[Any, Any, T_co]:
- """
- Call the given function with the given positional and keyword arguments.
- :return: the resulting coroutine
- :raises TypeError: if the return value was not a coroutine object
- """
- coro = func(*args, **kwargs)
- if not isinstance(coro, Coroutine):
- prefix = f"{func.__module__}." if hasattr(func, "__module__") else ""
- raise TypeError(
- f"Expected {prefix}{func.__qualname__}() to return a coroutine, but "
- f"the return value ({coro!r}) is not a coroutine object"
- )
- return coro
- class TaskStatus(Protocol[T_contra]):
- @overload
- def started(self: TaskStatus[None]) -> None: ...
- @overload
- def started(self, value: T_contra) -> None: ...
- def started(self, value: T_contra | None = None) -> None:
- """
- Signal that the task has started.
- :param value: object passed back to the starter of the task
- """
- class TaskGroup(metaclass=ABCMeta):
- """
- Groups several asynchronous tasks together.
- :ivar cancel_scope: the cancel scope inherited by all child tasks
- :vartype cancel_scope: CancelScope
- .. note:: On asyncio, support for eager task factories is considered to be
- **experimental**. In particular, they don't follow the usual semantics of new
- tasks being scheduled on the next iteration of the event loop, and may thus
- cause unexpected behavior in code that wasn't written with such semantics in
- mind.
- """
- cancel_scope: CancelScope
- def cancel(self, reason: str | None = None) -> None:
- """
- Cancel this task group's cancel scope immediately.
- This is a shortcut for calling ``.cancel_scope.cancel()`` on the task group.
- :param reason: a message describing the reason for the cancellation
- .. versionadded:: 4.14.0
- """
- self.cancel_scope.cancel(reason)
- @abstractmethod
- def create_task(
- self,
- coro: Coroutine[Any, Any, T_co],
- *,
- name: object = None,
- context: Context | None = None,
- ) -> TaskHandle[T_co]:
- """
- Create a new task from a coroutine object and schedule it to run.
- :param coro: a coroutine object
- :param name: optional name to give the task
- :param context: optional context to run the task in
- :return: a task handle
- .. versionadded:: 4.14.0
- """
- @final
- def start_soon(
- self,
- func: Callable[[Unpack[PosArgsT]], Coroutine[Any, Any, T_co]],
- *args: Unpack[PosArgsT],
- name: object = None,
- ) -> TaskHandle[T_co]:
- """
- Start a new task in this task group.
- :param func: a coroutine function
- :param args: positional arguments to call the function with
- :param name: name of the task, for the purposes of introspection and debugging
- :return: a task handle
- .. versionadded:: 3.0
- .. versionchanged:: 4.14.0
- This method now returns a task handle.
- """
- final_name = get_callable_name(func, name)
- return self.create_task(call_for_coroutine(func, args), name=final_name)
- @overload
- async def start(
- self,
- func: Callable[..., Coroutine[Any, Any, T_co]],
- *args: object,
- name: object = None,
- return_handle: Literal[False] = ...,
- ) -> Any: ...
- @overload
- async def start(
- self,
- func: Callable[..., Coroutine[Any, Any, T_co]],
- *args: object,
- name: object = None,
- return_handle: Literal[True],
- ) -> TaskHandle[T_co, Any]: ...
- @abstractmethod
- async def start(
- self,
- func: Callable[..., Coroutine[Any, Any, T_co]],
- *args: object,
- name: object = None,
- return_handle: Literal[False] | Literal[True] = False,
- ) -> Any:
- """
- Start a new task and wait until it signals for readiness.
- The target callable must accept a keyword argument ``task_status`` (of type
- :class:`TaskStatus`). Awaiting on this method will return whatever was passed to
- ``task_status.started()`` (``None`` by default).
- .. note:: The :class:`TaskStatus` class is generic, and the type argument should
- indicate the type of the value that will be passed to
- ``task_status.started()``.
- :param func: a coroutine function that accepts the ``task_status`` keyword
- argument
- :param args: positional arguments to call the function with
- :param name: an optional name for the task, for introspection and debugging
- :param return_handle: if ``True``, return a :class:`TaskHandle` which also
- contains the start value in ``start_value``
- :return: the value passed to ``task_status.started()``
- :raises RuntimeError: if the task finishes without calling
- ``task_status.started()``
- .. seealso:: :ref:`start_initialize`
- .. versionadded:: 3.0
- """
- @abstractmethod
- async def __aenter__(self) -> TaskGroup:
- """Enter the task group context and allow starting new tasks."""
- @abstractmethod
- async def __aexit__(
- self,
- exc_type: type[BaseException] | None,
- exc_val: BaseException | None,
- exc_tb: TracebackType | None,
- ) -> bool:
- """Exit the task group context waiting for all tasks to finish."""
|