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

1from __future__ import annotations 

2 

3import sys 

4from abc import ABCMeta, abstractmethod 

5from types import TracebackType 

6from typing import TYPE_CHECKING, Any, Awaitable, Callable, TypeVar, overload 

7from warnings import warn 

8 

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

10 from typing import Protocol 

11else: 

12 from typing_extensions import Protocol 

13 

14if TYPE_CHECKING: 

15 from anyio._core._tasks import CancelScope 

16 

17T_Retval = TypeVar("T_Retval") 

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

19 

20 

21class TaskStatus(Protocol[T_contra]): 

22 @overload 

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

24 ... 

25 

26 @overload 

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

28 ... 

29 

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

31 """ 

32 Signal that the task has started. 

33 

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

35 """ 

36 

37 

38class TaskGroup(metaclass=ABCMeta): 

39 """ 

40 Groups several asynchronous tasks together. 

41 

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

43 :vartype cancel_scope: CancelScope 

44 """ 

45 

46 cancel_scope: CancelScope 

47 

48 async def spawn( 

49 self, 

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

51 *args: object, 

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 .. deprecated:: 3.0 

62 Use :meth:`start_soon` instead. If your code needs AnyIO 2 compatibility, you 

63 can keep using this until AnyIO 4. 

64 

65 """ 

66 warn( 

67 'spawn() is deprecated -- use start_soon() (without the "await") instead', 

68 DeprecationWarning, 

69 ) 

70 self.start_soon(func, *args, name=name) 

71 

72 @abstractmethod 

73 def start_soon( 

74 self, 

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

76 *args: object, 

77 name: object = None, 

78 ) -> None: 

79 """ 

80 Start a new task in this task group. 

81 

82 :param func: a coroutine function 

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

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

85 

86 .. versionadded:: 3.0 

87 """ 

88 

89 @abstractmethod 

90 async def start( 

91 self, 

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

93 *args: object, 

94 name: object = None, 

95 ) -> Any: 

96 """ 

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

98 

99 :param func: a coroutine function 

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

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

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

103 :raises RuntimeError: if the task finishes without calling ``task_status.started()`` 

104 

105 .. versionadded:: 3.0 

106 """ 

107 

108 @abstractmethod 

109 async def __aenter__(self) -> TaskGroup: 

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

111 

112 @abstractmethod 

113 async def __aexit__( 

114 self, 

115 exc_type: type[BaseException] | None, 

116 exc_val: BaseException | None, 

117 exc_tb: TracebackType | None, 

118 ) -> bool | None: 

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