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

1from __future__ import annotations 

2 

3import sys 

4from abc import ABCMeta, abstractmethod 

5from collections.abc import Callable, Coroutine 

6from contextvars import Context 

7from types import TracebackType 

8from typing import TYPE_CHECKING, Any, Literal, Protocol, final, overload 

9 

10if sys.version_info >= (3, 13): 

11 from typing import TypeVar 

12else: 

13 from typing_extensions import TypeVar 

14 

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

16 from typing import TypeVarTuple, Unpack 

17else: 

18 from typing_extensions import TypeVarTuple, Unpack 

19 

20if TYPE_CHECKING: 

21 from .._core._tasks import CancelScope, TaskHandle 

22 

23T_co = TypeVar("T_co", covariant=True) 

24T_contra = TypeVar("T_contra", contravariant=True, default=None) 

25PosArgsT = TypeVarTuple("PosArgsT") 

26 

27 

28def get_callable_name(func: Callable, override: object = None) -> str: 

29 if override is not None: 

30 return str(override) 

31 

32 module = getattr(func, "__module__", None) 

33 qualname = getattr(func, "__qualname__", None) 

34 return ".".join([x for x in (module, qualname) if x]) 

35 

36 

37def call_for_coroutine( 

38 func: Callable[[Unpack[PosArgsT]], Coroutine[Any, Any, T_co]], 

39 args: tuple[Unpack[PosArgsT]], 

40 **kwargs: Any, 

41) -> Coroutine[Any, Any, T_co]: 

42 """ 

43 Call the given function with the given positional and keyword arguments. 

44 

45 :return: the resulting coroutine 

46 :raises TypeError: if the return value was not a coroutine object 

47 

48 """ 

49 coro = func(*args, **kwargs) 

50 if not isinstance(coro, Coroutine): 

51 prefix = f"{func.__module__}." if hasattr(func, "__module__") else "" 

52 raise TypeError( 

53 f"Expected {prefix}{func.__qualname__}() to return a coroutine, but " 

54 f"the return value ({coro!r}) is not a coroutine object" 

55 ) 

56 

57 return coro 

58 

59 

60class TaskStatus(Protocol[T_contra]): 

61 @overload 

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

63 

64 @overload 

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

66 

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

68 """ 

69 Signal that the task has started. 

70 

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

72 """ 

73 

74 

75class TaskGroup(metaclass=ABCMeta): 

76 """ 

77 Groups several asynchronous tasks together. 

78 

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

80 :vartype cancel_scope: CancelScope 

81 

82 .. note:: On asyncio, support for eager task factories is considered to be 

83 **experimental**. In particular, they don't follow the usual semantics of new 

84 tasks being scheduled on the next iteration of the event loop, and may thus 

85 cause unexpected behavior in code that wasn't written with such semantics in 

86 mind. 

87 """ 

88 

89 cancel_scope: CancelScope 

90 

91 def cancel(self, reason: str | None = None) -> None: 

92 """ 

93 Cancel this task group's cancel scope immediately. 

94 

95 This is a shortcut for calling ``.cancel_scope.cancel()`` on the task group. 

96 

97 :param reason: a message describing the reason for the cancellation 

98 

99 .. versionadded:: 4.14.0 

100 

101 """ 

102 self.cancel_scope.cancel(reason) 

103 

104 @abstractmethod 

105 def create_task( 

106 self, 

107 coro: Coroutine[Any, Any, T_co], 

108 *, 

109 name: object = None, 

110 context: Context | None = None, 

111 ) -> TaskHandle[T_co]: 

112 """ 

113 Create a new task from a coroutine object and schedule it to run. 

114 

115 :param coro: a coroutine object 

116 :param name: optional name to give the task 

117 :param context: optional context to run the task in 

118 :return: a task handle 

119 

120 .. versionadded:: 4.14.0 

121 """ 

122 

123 @final 

124 def start_soon( 

125 self, 

126 func: Callable[[Unpack[PosArgsT]], Coroutine[Any, Any, T_co]], 

127 *args: Unpack[PosArgsT], 

128 name: object = None, 

129 ) -> TaskHandle[T_co]: 

130 """ 

131 Start a new task in this task group. 

132 

133 :param func: a coroutine function 

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

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

136 :return: a task handle 

137 

138 .. versionadded:: 3.0 

139 .. versionchanged:: 4.14.0 

140 This method now returns a task handle. 

141 

142 """ 

143 final_name = get_callable_name(func, name) 

144 return self.create_task(call_for_coroutine(func, args), name=final_name) 

145 

146 @overload 

147 async def start( 

148 self, 

149 func: Callable[..., Coroutine[Any, Any, T_co]], 

150 *args: object, 

151 name: object = None, 

152 return_handle: Literal[False] = ..., 

153 ) -> Any: ... 

154 

155 @overload 

156 async def start( 

157 self, 

158 func: Callable[..., Coroutine[Any, Any, T_co]], 

159 *args: object, 

160 name: object = None, 

161 return_handle: Literal[True], 

162 ) -> TaskHandle[T_co, Any]: ... 

163 

164 @abstractmethod 

165 async def start( 

166 self, 

167 func: Callable[..., Coroutine[Any, Any, T_co]], 

168 *args: object, 

169 name: object = None, 

170 return_handle: Literal[False] | Literal[True] = False, 

171 ) -> Any: 

172 """ 

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

174 

175 The target callable must accept a keyword argument ``task_status`` (of type 

176 :class:`TaskStatus`). Awaiting on this method will return whatever was passed to 

177 ``task_status.started()`` (``None`` by default). 

178 

179 .. note:: The :class:`TaskStatus` class is generic, and the type argument should 

180 indicate the type of the value that will be passed to 

181 ``task_status.started()``. 

182 

183 :param func: a coroutine function that accepts the ``task_status`` keyword 

184 argument 

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

186 :param name: an optional name for the task, for introspection and debugging 

187 :param return_handle: if ``True``, return a :class:`TaskHandle` which also 

188 contains the start value in ``start_value`` 

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

190 :raises RuntimeError: if the task finishes without calling 

191 ``task_status.started()`` 

192 

193 .. seealso:: :ref:`start_initialize` 

194 

195 .. versionadded:: 3.0 

196 """ 

197 

198 @abstractmethod 

199 async def __aenter__(self) -> TaskGroup: 

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

201 

202 @abstractmethod 

203 async def __aexit__( 

204 self, 

205 exc_type: type[BaseException] | None, 

206 exc_val: BaseException | None, 

207 exc_tb: TracebackType | None, 

208 ) -> bool: 

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