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

1import typing 

2from abc import ABCMeta, abstractmethod 

3from types import TracebackType 

4from typing import Any, Callable, Coroutine, Optional, Type, TypeVar 

5from warnings import warn 

6 

7if typing.TYPE_CHECKING: 

8 from anyio._core._tasks import CancelScope 

9 

10T_Retval = TypeVar("T_Retval") 

11 

12 

13class TaskStatus(metaclass=ABCMeta): 

14 @abstractmethod 

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

16 """ 

17 Signal that the task has started. 

18 

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

20 """ 

21 

22 

23class TaskGroup(metaclass=ABCMeta): 

24 """ 

25 Groups several asynchronous tasks together. 

26 

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

28 :vartype cancel_scope: CancelScope 

29 """ 

30 

31 cancel_scope: "CancelScope" 

32 

33 async def spawn( 

34 self, 

35 func: Callable[..., Coroutine[Any, Any, Any]], 

36 *args: object, 

37 name: object = None 

38 ) -> None: 

39 """ 

40 Start a new task in this task group. 

41 

42 :param func: a coroutine function 

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

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

45 

46 .. deprecated:: 3.0 

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

48 can keep using this until AnyIO 4. 

49 

50 """ 

51 warn( 

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

53 DeprecationWarning, 

54 ) 

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

56 

57 @abstractmethod 

58 def start_soon( 

59 self, 

60 func: Callable[..., Coroutine[Any, Any, Any]], 

61 *args: object, 

62 name: object = None 

63 ) -> None: 

64 """ 

65 Start a new task in this task group. 

66 

67 :param func: a coroutine function 

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

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

70 

71 .. versionadded:: 3.0 

72 """ 

73 

74 @abstractmethod 

75 async def start( 

76 self, 

77 func: Callable[..., Coroutine[Any, Any, Any]], 

78 *args: object, 

79 name: object = None 

80 ) -> object: 

81 """ 

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

83 

84 :param func: a coroutine function 

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

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

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

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

89 

90 .. versionadded:: 3.0 

91 """ 

92 

93 @abstractmethod 

94 async def __aenter__(self) -> "TaskGroup": 

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

96 

97 @abstractmethod 

98 async def __aexit__( 

99 self, 

100 exc_type: Optional[Type[BaseException]], 

101 exc_val: Optional[BaseException], 

102 exc_tb: Optional[TracebackType], 

103 ) -> Optional[bool]: 

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