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

1from __future__ import annotations 

2 

3import math 

4import sys 

5from collections.abc import ( 

6 Coroutine, 

7 Generator, 

8) 

9from contextlib import contextmanager 

10from enum import Enum, auto 

11from inspect import iscoroutine 

12from types import TracebackType 

13from typing import Any, Generic, final 

14 

15from ..abc import TaskGroup, TaskStatus 

16from ._eventloop import get_async_backend, get_cancelled_exc_class 

17from ._exceptions import TaskCancelled, TaskFailed, TaskNotFinished 

18 

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

20 from typing import TypeVar 

21else: 

22 from typing_extensions import TypeVar 

23 

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

25 from typing import Never, Self, TypeVarTuple 

26else: 

27 from typing_extensions import Never, Self, TypeVarTuple 

28 

29T = TypeVar("T") 

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

31T_startval_co = TypeVar("T_startval_co", covariant=True, default=Never) 

32PosArgsT = TypeVarTuple("PosArgsT") 

33 

34 

35class _IgnoredTaskStatus(TaskStatus[object]): 

36 def started(self, value: object = None) -> None: 

37 pass 

38 

39 

40TASK_STATUS_IGNORED = _IgnoredTaskStatus() 

41 

42 

43class CancelScope: 

44 """ 

45 Wraps a unit of work that can be made separately cancellable. 

46 

47 :param deadline: The time (clock value) when this scope is cancelled automatically 

48 :param shield: ``True`` to shield the cancel scope from external cancellation 

49 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

50 current thread 

51 """ 

52 

53 __slots__ = ("__weakref__",) 

54 

55 def __new__( 

56 cls, *, deadline: float = math.inf, shield: bool = False 

57 ) -> CancelScope: 

58 return get_async_backend().create_cancel_scope(shield=shield, deadline=deadline) 

59 

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

61 """ 

62 Cancel this scope immediately. 

63 

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

65 

66 """ 

67 raise NotImplementedError 

68 

69 @property 

70 def deadline(self) -> float: 

71 """ 

72 The time (clock value) when this scope is cancelled automatically. 

73 

74 Will be ``float('inf')`` if no timeout has been set. 

75 

76 """ 

77 raise NotImplementedError 

78 

79 @deadline.setter 

80 def deadline(self, value: float) -> None: 

81 raise NotImplementedError 

82 

83 @property 

84 def cancel_called(self) -> bool: 

85 """``True`` if :meth:`cancel` has been called.""" 

86 raise NotImplementedError 

87 

88 @property 

89 def cancelled_caught(self) -> bool: 

90 """ 

91 ``True`` if this scope suppressed a cancellation exception it itself raised. 

92 

93 This is typically used to check if any work was interrupted, or to see if the 

94 scope was cancelled due to its deadline being reached. The value will, however, 

95 only be ``True`` if the cancellation was triggered by the scope itself (and not 

96 an outer scope). 

97 

98 """ 

99 raise NotImplementedError 

100 

101 @property 

102 def shield(self) -> bool: 

103 """ 

104 ``True`` if this scope is shielded from external cancellation. 

105 

106 While a scope is shielded, it will not receive cancellations from outside. 

107 

108 """ 

109 raise NotImplementedError 

110 

111 @shield.setter 

112 def shield(self, value: bool) -> None: 

113 raise NotImplementedError 

114 

115 def __enter__(self) -> Self: 

116 raise NotImplementedError 

117 

118 def __exit__( 

119 self, 

120 exc_type: type[BaseException] | None, 

121 exc_val: BaseException | None, 

122 exc_tb: TracebackType | None, 

123 ) -> bool: 

124 raise NotImplementedError 

125 

126 

127@contextmanager 

128def fail_at( 

129 deadline: float | None, shield: bool = False, reason: str | None = None 

130) -> Generator[CancelScope, None, None]: 

131 """ 

132 Create a context manager which raises a :class:`TimeoutError` if the code in the 

133 enclosing context does not finish by the given deadline. 

134 

135 :param deadline: the deadline before raising the exception, or 

136 ``None`` to disable the timeout 

137 :param shield: ``True`` to shield the cancel scope from external cancellation 

138 :param reason: explanation for timeout to add to the message of a raised `TimeoutError` 

139 :return: a context manager that yields a cancel scope 

140 :rtype: :class:`~typing.ContextManager`\\[:class:`~anyio.CancelScope`\\] 

141 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

142 current thread 

143 

144 .. versionadded:: 4.15.0 

145 

146 """ 

147 current_time = get_async_backend().current_time 

148 effective_deadline = math.inf if deadline is None else deadline 

149 with get_async_backend().create_cancel_scope( 

150 deadline=effective_deadline, shield=shield 

151 ) as cancel_scope: 

152 yield cancel_scope 

153 

154 if cancel_scope.cancelled_caught and current_time() >= cancel_scope.deadline: 

155 raise TimeoutError(reason) if reason else TimeoutError 

156 

157 

158@contextmanager 

159def fail_after( 

160 delay: float | None, shield: bool = False, reason: str | None = None 

161) -> Generator[CancelScope, None, None]: 

162 """ 

163 Create a context manager which raises a :class:`TimeoutError` if the code in the 

164 enclosing context block does not finish in time. 

165 

166 :param delay: maximum allowed time (in seconds) before raising the exception, or 

167 ``None`` to disable the timeout 

168 :param shield: ``True`` to shield the cancel scope from external cancellation 

169 :param reason: explanation for timeout to add to the message of a raised `TimeoutError` 

170 :return: a context manager that yields a cancel scope 

171 :rtype: :class:`~typing.ContextManager`\\[:class:`~anyio.CancelScope`\\] 

172 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

173 current thread 

174 

175 """ 

176 current_time = get_async_backend().current_time 

177 deadline = (current_time() + delay) if delay is not None else math.inf 

178 with fail_at(deadline, shield=shield, reason=reason) as scope: 

179 yield scope 

180 

181 

182def move_on_at(deadline: float | None, shield: bool = False) -> CancelScope: 

183 """ 

184 Create a cancel scope with a deadline that expires after the given delay. 

185 

186 :param deadline: the deadline before exiting the context block, or 

187 ``None`` to disable the timeout 

188 :param shield: ``True`` to shield the cancel scope from external cancellation 

189 :return: a cancel scope 

190 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

191 current thread 

192 

193 .. versionadded:: 4.15.0 

194 

195 """ 

196 return get_async_backend().create_cancel_scope( 

197 deadline=deadline if deadline is not None else math.inf, shield=shield 

198 ) 

199 

200 

201def move_on_after(delay: float | None, shield: bool = False) -> CancelScope: 

202 """ 

203 Create a cancel scope with a deadline that expires after the given delay. 

204 

205 :param delay: maximum allowed time (in seconds) before exiting the context block, or 

206 ``None`` to disable the timeout 

207 :param shield: ``True`` to shield the cancel scope from external cancellation 

208 :return: a cancel scope 

209 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

210 current thread 

211 

212 .. note:: Unlike with :func:`fail_after`, the timer starts when this function is 

213 called, not when the context manager is entered. This will be changed in v5.0. 

214 

215 """ 

216 deadline = ( 

217 (get_async_backend().current_time() + delay) if delay is not None else math.inf 

218 ) 

219 return get_async_backend().create_cancel_scope(deadline=deadline, shield=shield) 

220 

221 

222def current_effective_deadline() -> float: 

223 """ 

224 Return the nearest deadline among all the cancel scopes effective for the current 

225 task. 

226 

227 :return: a clock value from the event loop's internal clock (or ``float('inf')`` if 

228 there is no deadline in effect, or ``float('-inf')`` if the current scope has 

229 been cancelled) 

230 :rtype: float 

231 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

232 current thread 

233 

234 """ 

235 return get_async_backend().current_effective_deadline() 

236 

237 

238def create_task_group() -> TaskGroup: 

239 """ 

240 Create a task group. 

241 

242 :return: a task group 

243 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

244 current thread 

245 

246 """ 

247 return get_async_backend().create_task_group() 

248 

249 

250@final 

251class TaskHandle(Generic[T_co, T_startval_co]): 

252 """ 

253 Returned from the task-spawning methods of :class:`TaskGroup`. Can be awaited on to 

254 get the return value of the task (or the raised exception). If the task was 

255 terminated by a :exc:`BaseException`, :exc:`TaskFailed` will be raised (or its 

256 subclass :exc:`TaskCancelled` if the task was cancelled). 

257 

258 .. versionadded:: 4.14.0 

259 """ 

260 

261 class Status(Enum): 

262 """ 

263 The status of a task handle. 

264 

265 .. attribute:: PENDING 

266 

267 The task has not finished yet. 

268 .. attribute:: FINISHED 

269 

270 The task has finished with a return value. 

271 .. attribute:: CANCELLING 

272 

273 The task has been cancelled but has not finished yet. 

274 .. attribute:: CANCELLED 

275 

276 The task was cancelled and has finished since. 

277 .. attribute:: FAILED 

278 

279 The task raised an exception. 

280 """ 

281 

282 PENDING = auto() 

283 FINISHED = auto() 

284 CANCELLING = auto() 

285 CANCELLED = auto() 

286 FAILED = auto() 

287 

288 __slots__ = ( 

289 "__weakref__", 

290 "_cancel_scope", 

291 "_coro", 

292 "_exception", 

293 "_finished_event", 

294 "_name", 

295 "_return_value", 

296 "_start_value", 

297 ) 

298 

299 _return_value: T_co 

300 _start_value: T_startval_co 

301 

302 def __init__( 

303 self, 

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

305 name: object, 

306 cancel_scope: CancelScope | None = None, 

307 ) -> None: 

308 from ._synchronization import Event 

309 

310 self._coro = coro 

311 self._cancel_scope = cancel_scope if cancel_scope is not None else CancelScope() 

312 self._finished_event = Event() 

313 self._exception: BaseException | None = None 

314 

315 if name is not None: 

316 self._name = str(name) 

317 elif iscoroutine(coro): 

318 self._name = coro.__qualname__ 

319 else: 

320 self._name = str(coro) # coroutine-like object (e.g. asend() objects) 

321 

322 async def _run_coro(self) -> None: 

323 __tracebackhide__ = True 

324 

325 with self._cancel_scope: 

326 try: 

327 retval = await self._coro 

328 except BaseException as exc: 

329 self._exception = exc 

330 raise 

331 else: 

332 self._return_value = retval 

333 finally: 

334 self._finished_event.set() 

335 del self # Break the reference cycle 

336 

337 def cancel(self) -> None: 

338 """ 

339 Set the task to a cancelled state. 

340 

341 This will interrupt any interruptible asynchronous operation, and will cause 

342 any further awaits on this task to get immediately cancelled, unless done in 

343 a shielded cancel scope. 

344 

345 If the task has already finished, this method has no effect. 

346 """ 

347 if not self._finished_event.is_set(): 

348 self._cancel_scope.cancel() 

349 

350 @property 

351 def coro(self) -> Coroutine[Any, Any, T_co]: 

352 """ 

353 The coroutine object that was passed to one of the task-spawning methods in 

354 :class:`TaskGroup`. 

355 """ 

356 return self._coro 

357 

358 @property 

359 def status(self) -> TaskHandle.Status: 

360 """ 

361 The current status of the task. 

362 

363 Every task starts in the :attr:`~TaskHandle.Status.PENDING` state. 

364 If a task is cancelled while in this state, it will transition to the 

365 :attr:`~TaskHandle.Status.CANCELLING` state. When the task finishes, it will 

366 transition to one of the three final states ( 

367 :attr:`~TaskHandle.Status.FINISHED`, :attr:`~TaskHandle.Status.FAILED`, or 

368 :attr:`~TaskHandle.Status.CANCELLING`) depending on the exception the task 

369 raised, if any. No other status transitions will happen. 

370 """ 

371 if not self._finished_event.is_set(): 

372 if self._cancel_scope.cancel_called: 

373 return TaskHandle.Status.CANCELLING 

374 else: 

375 return TaskHandle.Status.PENDING 

376 elif self._exception is not None: 

377 if isinstance(self._exception, get_cancelled_exc_class()): 

378 return TaskHandle.Status.CANCELLED 

379 else: 

380 return TaskHandle.Status.FAILED 

381 else: 

382 return TaskHandle.Status.FINISHED 

383 

384 @property 

385 def name(self) -> str: 

386 """The name of the task.""" 

387 return self._name 

388 

389 @property 

390 def exception(self) -> BaseException | None: 

391 """ 

392 The exception raised by the task, or ``None`` if it finished without raising. 

393 

394 :raises TaskNotFinished: if the task has not finished yet 

395 :raises TaskCancelled: if the task was cancelled 

396 

397 """ 

398 match self.status: 

399 case TaskHandle.Status.PENDING: 

400 raise TaskNotFinished("the task has not finished yet") 

401 case TaskHandle.Status.FINISHED: 

402 return None 

403 case TaskHandle.Status.CANCELLING: 

404 raise TaskCancelled("the task was cancelled") 

405 case TaskHandle.Status.CANCELLED: 

406 raise TaskCancelled("the task was cancelled") from self._exception 

407 case TaskHandle.Status.FAILED: 

408 return self._exception 

409 

410 @property 

411 def return_value(self) -> T_co: 

412 """ 

413 The return value of the task. 

414 

415 :raises TaskNotFinished: if the task has not finished yet 

416 :raises TaskCancelled: if the task was cancelled 

417 :raises TaskFailed: if the task raised an exception 

418 

419 """ 

420 match self.status: 

421 case TaskHandle.Status.PENDING: 

422 raise TaskNotFinished("the task has not finished yet") 

423 case TaskHandle.Status.FINISHED: 

424 return self._return_value 

425 case TaskHandle.Status.CANCELLING: 

426 raise TaskCancelled("the task was cancelled") 

427 case TaskHandle.Status.CANCELLED: 

428 raise TaskCancelled("the task was cancelled") from self._exception 

429 case TaskHandle.Status.FAILED: 

430 raise TaskFailed("the task raised an exception") from self._exception 

431 

432 @property 

433 def start_value(self) -> T_startval_co: 

434 """ 

435 The value passed to :meth:`task_status.started() <.abc.TaskStatus.started>`, 

436 

437 :raises RuntimeError: if the task was not started with :meth:`TaskGroup.start() 

438 <.abc.TaskGroup.start>` 

439 """ 

440 try: 

441 return self._start_value 

442 except AttributeError: 

443 raise RuntimeError( 

444 "the task was not started with TaskGroup.start()" 

445 ) from None 

446 

447 async def wait(self) -> None: 

448 """ 

449 Wait for the task to finish. 

450 

451 This method will return as soon as the task has finished, no matter how it 

452 happened. 

453 """ 

454 await self._finished_event.wait() 

455 

456 def __await__(self) -> Generator[Any, Any, T_co]: 

457 yield from self._finished_event.wait().__await__() 

458 return self.return_value 

459 

460 def __repr__(self) -> str: 

461 return ( 

462 f"<{self.__class__.__name__} {self.status.name.lower()} " 

463 f"name={self._name!r} coro={self._coro!r}>" 

464 )