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

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 

27 @overload 

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

29 ... 

30 

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

32 """ 

33 Signal that the task has started. 

34 

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

36 """ 

37 

38 

39class TaskGroup(metaclass=ABCMeta): 

40 """ 

41 Groups several asynchronous tasks together. 

42 

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

44 :vartype cancel_scope: CancelScope 

45 """ 

46 

47 cancel_scope: CancelScope 

48 

49 @abstractmethod 

50 def start_soon( 

51 self, 

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

53 *args: Unpack[PosArgsT], 

54 name: object = None, 

55 ) -> None: 

56 """ 

57 Start a new task in this task group. 

58 

59 :param func: a coroutine function 

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

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

62 

63 .. versionadded:: 3.0 

64 """ 

65 

66 @abstractmethod 

67 async def start( 

68 self, 

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

70 *args: object, 

71 name: object = None, 

72 ) -> Any: 

73 """ 

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

75 

76 :param func: a coroutine function 

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

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

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

80 :raises RuntimeError: if the task finishes without calling 

81 ``task_status.started()`` 

82 

83 .. versionadded:: 3.0 

84 """ 

85 

86 @abstractmethod 

87 async def __aenter__(self) -> TaskGroup: 

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

89 

90 @abstractmethod 

91 async def __aexit__( 

92 self, 

93 exc_type: type[BaseException] | None, 

94 exc_val: BaseException | None, 

95 exc_tb: TracebackType | None, 

96 ) -> bool | None: 

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