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

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_coro_name(coro: Coroutine[Any, Any, object], override: object = None) -> str: 

29 if override is not None: 

30 return str(override) 

31 

32 try: 

33 cr_frame = coro.cr_frame # type: ignore[attr-defined] 

34 module = cr_frame.f_globals["__name__"] 

35 except (AttributeError, KeyError): 

36 module = None 

37 

38 qualname = getattr(coro, "__qualname__", None) 

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

40 

41 

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

43 if override is not None: 

44 return str(override) 

45 

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

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

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

49 

50 

51def call_for_coroutine( 

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

53 args: tuple[Unpack[PosArgsT]], 

54 **kwargs: Any, 

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

56 """ 

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

58 

59 :return: the resulting coroutine 

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

61 

62 """ 

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

64 if not isinstance(coro, Coroutine): 

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

66 raise TypeError( 

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

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

69 ) 

70 

71 return coro 

72 

73 

74class TaskStatus(Protocol[T_contra]): 

75 @overload 

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

77 

78 @overload 

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

80 

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

82 """ 

83 Signal that the task has started. 

84 

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

86 """ 

87 

88 

89class TaskGroup(metaclass=ABCMeta): 

90 """ 

91 Groups several asynchronous tasks together. 

92 

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

94 :vartype cancel_scope: CancelScope 

95 

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

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

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

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

100 mind. 

101 """ 

102 

103 cancel_scope: CancelScope 

104 

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

106 """ 

107 Cancel this task group's cancel scope immediately. 

108 

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

110 

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

112 

113 .. versionadded:: 4.14.0 

114 

115 """ 

116 self.cancel_scope.cancel(reason) 

117 

118 @abstractmethod 

119 def create_task( 

120 self, 

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

122 *, 

123 name: object = None, 

124 context: Context | None = None, 

125 ) -> TaskHandle[T_co]: 

126 """ 

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

128 

129 :param coro: a coroutine object 

130 :param name: optional name to give the task 

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

132 :return: a task handle 

133 

134 .. versionadded:: 4.14.0 

135 """ 

136 

137 @final 

138 def start_soon( 

139 self, 

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

141 *args: Unpack[PosArgsT], 

142 name: object = None, 

143 ) -> TaskHandle[T_co]: 

144 """ 

145 Start a new task in this task group. 

146 

147 :param func: a coroutine function 

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

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

150 :return: a task handle 

151 

152 .. versionadded:: 3.0 

153 .. versionchanged:: 4.14.0 

154 This method now returns a task handle. 

155 

156 """ 

157 final_name = get_callable_name(func, name) 

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

159 

160 @overload 

161 async def start( 

162 self, 

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

164 *args: object, 

165 name: object = None, 

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

167 ) -> Any: ... 

168 

169 @overload 

170 async def start( 

171 self, 

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

173 *args: object, 

174 name: object = None, 

175 return_handle: Literal[True], 

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

177 

178 @abstractmethod 

179 async def start( 

180 self, 

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

182 *args: object, 

183 name: object = None, 

184 return_handle: Literal[False, True] = False, 

185 ) -> Any: 

186 """ 

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

188 

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

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

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

192 

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

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

195 ``task_status.started()``. 

196 

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

198 argument 

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

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

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

202 contains the start value in ``start_value`` 

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

204 :raises RuntimeError: if the task finishes without calling 

205 ``task_status.started()`` 

206 

207 .. seealso:: :ref:`start_initialize` 

208 

209 .. versionadded:: 3.0 

210 """ 

211 

212 @abstractmethod 

213 async def __aenter__(self) -> TaskGroup: 

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

215 

216 @abstractmethod 

217 async def __aexit__( 

218 self, 

219 exc_type: type[BaseException] | None, 

220 exc_val: BaseException | None, 

221 exc_tb: TracebackType | None, 

222 ) -> bool: 

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