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

1from __future__ import annotations 

2 

3from abc import ABCMeta, abstractmethod 

4from collections.abc import Awaitable, Callable 

5from types import TracebackType 

6from typing import TYPE_CHECKING, Any, Protocol, TypeVar, overload 

7 

8if TYPE_CHECKING: 

9 from .._core._tasks import CancelScope 

10 

11T_Retval = TypeVar("T_Retval") 

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

13 

14 

15class TaskStatus(Protocol[T_contra]): 

16 @overload 

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

18 ... 

19 

20 @overload 

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

22 ... 

23 

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

25 """ 

26 Signal that the task has started. 

27 

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

29 """ 

30 

31 

32class TaskGroup(metaclass=ABCMeta): 

33 """ 

34 Groups several asynchronous tasks together. 

35 

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

37 :vartype cancel_scope: CancelScope 

38 """ 

39 

40 cancel_scope: CancelScope 

41 

42 @abstractmethod 

43 def start_soon( 

44 self, 

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

46 *args: object, 

47 name: object = None, 

48 ) -> None: 

49 """ 

50 Start a new task in this task group. 

51 

52 :param func: a coroutine function 

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

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

55 

56 .. versionadded:: 3.0 

57 """ 

58 

59 @abstractmethod 

60 async def start( 

61 self, 

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

63 *args: object, 

64 name: object = None, 

65 ) -> Any: 

66 """ 

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

68 

69 :param func: a coroutine function 

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

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

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

73 :raises RuntimeError: if the task finishes without calling 

74 ``task_status.started()`` 

75 

76 .. versionadded:: 3.0 

77 """ 

78 

79 @abstractmethod 

80 async def __aenter__(self) -> TaskGroup: 

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

82 

83 @abstractmethod 

84 async def __aexit__( 

85 self, 

86 exc_type: type[BaseException] | None, 

87 exc_val: BaseException | None, 

88 exc_tb: TracebackType | None, 

89 ) -> bool | None: 

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