Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/anyio/abc/_tasks.py: 93%

1from __future__ import annotations 

2 

3import sys 

4from abc import ABCMeta, abstractmethod 

5from collections.abc import Awaitable, Callable 

6from types import TracebackType 

7from typing import TYPE_CHECKING, Any, Protocol, TypeVar, overload 

8 

9if sys.version_info >= (3, 11): 

10 from typing import TypeVarTuple, Unpack 

11else: 

12 from typing_extensions import TypeVarTuple, Unpack 

13 

14if TYPE_CHECKING: 

15 from .._core._tasks import CancelScope 

16 

17T_Retval = TypeVar("T_Retval") 

18T_contra = TypeVar("T_contra", contravariant=True) 

19PosArgsT = TypeVarTuple("PosArgsT") 

20 

21 

22class TaskStatus(Protocol[T_contra]): 

23 @overload 

24 def started(self: TaskStatus[None]) -> None: ... 

25 

26 @overload 

27 def started(self, value: T_contra) -> None: ... 

28 

29 def started(self, value: T_contra | None = None) -> None: 

30 """ 

31 Signal that the task has started. 

32 

33 :param value: object passed back to the starter of the task 

34 """ 

35 

36 

37class TaskGroup(metaclass=ABCMeta): 

38 """ 

39 Groups several asynchronous tasks together. 

40 

41 :ivar cancel_scope: the cancel scope inherited by all child tasks 

42 :vartype cancel_scope: CancelScope 

43 """ 

44 

45 cancel_scope: CancelScope 

46 

47 @abstractmethod 

48 def start_soon( 

49 self, 

50 func: Callable[[Unpack[PosArgsT]], Awaitable[Any]], 

51 *args: Unpack[PosArgsT], 

52 name: object = None, 

53 ) -> None: 

54 """ 

55 Start a new task in this task group. 

56 

57 :param func: a coroutine function 

58 :param args: positional arguments to call the function with 

59 :param name: name of the task, for the purposes of introspection and debugging 

60 

61 .. versionadded:: 3.0 

62 """ 

63 

64 @abstractmethod 

65 async def start( 

66 self, 

67 func: Callable[..., Awaitable[Any]], 

68 *args: object, 

69 name: object = None, 

70 ) -> Any: 

71 """ 

72 Start a new task and wait until it signals for readiness. 

73 

74 :param func: a coroutine function 

75 :param args: positional arguments to call the function with 

76 :param name: name of the task, for the purposes of introspection and debugging 

77 :return: the value passed to ``task_status.started()`` 

78 :raises RuntimeError: if the task finishes without calling 

79 ``task_status.started()`` 

80 

81 .. versionadded:: 3.0 

82 """ 

83 

84 @abstractmethod 

85 async def __aenter__(self) -> TaskGroup: 

86 """Enter the task group context and allow starting new tasks.""" 

87 

88 @abstractmethod 

89 async def __aexit__( 

90 self, 

91 exc_type: type[BaseException] | None, 

92 exc_val: BaseException | None, 

93 exc_tb: TracebackType | None, 

94 ) -> bool | None: 

95 """Exit the task group context waiting for all tasks to finish."""