Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pyee/asyncio.py: 38%

1# -*- coding: utf-8 -*- 

2 

3from asyncio import AbstractEventLoop, ensure_future, Future, iscoroutine, wait 

4from typing import Any, Callable, cast, Dict, Optional, Set, Tuple 

5 

6from pyee.base import EventEmitter 

7 

8Self = Any 

9 

10__all__ = ["AsyncIOEventEmitter"] 

11 

12 

13class AsyncIOEventEmitter(EventEmitter): 

14 """An event emitter class which can run asyncio coroutines in addition to 

15 synchronous blocking functions. For example: 

16 

17 ```py 

18 @ee.on('event') 

19 async def async_handler(*args, **kwargs): 

20 await returns_a_future() 

21 ``` 

22 

23 On emit, the event emitter will automatically schedule the coroutine using 

24 `asyncio.ensure_future` and the configured event loop (defaults to 

25 `asyncio.get_event_loop()`). 

26 

27 Unlike the case with the EventEmitter, all exceptions raised by 

28 event handlers are automatically emitted on the `error` event. This is 

29 important for asyncio coroutines specifically but is also handled for 

30 synchronous functions for consistency. 

31 

32 When `loop` is specified, the supplied event loop will be used when 

33 scheduling work with `ensure_future`. Otherwise, the default asyncio 

34 event loop is used. 

35 

36 For asyncio coroutine event handlers, calling emit is non-blocking. 

37 In other words, you do not have to await any results from emit, and the 

38 coroutine is scheduled in a fire-and-forget fashion. 

39 """ 

40 

41 def __init__(self: Self, loop: Optional[AbstractEventLoop] = None) -> None: 

42 super(AsyncIOEventEmitter, self).__init__() 

43 self._loop: Optional[AbstractEventLoop] = loop 

44 self._waiting: Set[Future] = set() 

45 

46 def emit( 

47 self: Self, 

48 event: str, 

49 *args: Any, 

50 **kwargs: Any, 

51 ) -> bool: 

52 """Emit `event`, passing `*args` and `**kwargs` to each attached 

53 function or coroutine. Returns `True` if any functions are attached to 

54 `event`; otherwise returns `False`. 

55 

56 Example: 

57 

58 ```py 

59 ee.emit('data', '00101001') 

60 ``` 

61 

62 Assuming `data` is an attached function, this will call 

63 `data('00101001')'`. 

64 

65 When executing coroutine handlers, their respective futures will be 

66 stored in a "waiting" state. These futures may be waited on or 

67 canceled with `wait_for_complete` and `cancel`, respectively; and 

68 their status may be checked via the `complete` property. 

69 """ 

70 return super().emit(event, *args, **kwargs) 

71 

72 def _emit_run( 

73 self: Self, 

74 f: Callable, 

75 args: Tuple[Any, ...], 

76 kwargs: Dict[str, Any], 

77 ) -> None: 

78 try: 

79 coro: Any = f(*args, **kwargs) 

80 except Exception as exc: 

81 self.emit("error", exc) 

82 else: 

83 if iscoroutine(coro): 

84 if self._loop: 

85 # ensure_future is *extremely* cranky about the types here, 

86 # but this is relatively well-tested and I think the types 

87 # are more strict than they should be 

88 fut: Any = ensure_future(cast(Any, coro), loop=self._loop) 

89 else: 

90 fut = ensure_future(cast(Any, coro)) 

91 

92 elif isinstance(coro, Future): 

93 fut = cast(Any, coro) 

94 else: 

95 return 

96 

97 def callback(f: Future) -> None: 

98 self._waiting.discard(f) 

99 

100 if f.cancelled(): 

101 return 

102 

103 exc: Optional[BaseException] = f.exception() 

104 if exc: 

105 self.emit("error", exc) 

106 

107 fut.add_done_callback(callback) 

108 self._waiting.add(fut) 

109 

110 async def wait_for_complete(self: Self) -> None: 

111 """Waits for all pending tasks to complete. For example: 

112 

113 ```py 

114 @ee.on('event') 

115 async def async_handler(*args, **kwargs): 

116 await returns_a_future() 

117 

118 # Triggers execution of async_handler 

119 ee.emit('data', '00101001') 

120 

121 await ee.wait_for_complete() 

122 

123 # async_handler has completed execution 

124 ``` 

125 

126 This is useful if you're attempting a graceful shutdown of your 

127 application and want to ensure all coroutines have completed execution 

128 beforehand. 

129 """ 

130 if self._waiting: 

131 await wait(self._waiting) 

132 

133 def cancel(self: Self) -> None: 

134 """Cancel all pending tasks. For example: 

135 

136 ```py 

137 @ee.on('event') 

138 async def async_handler(*args, **kwargs): 

139 await returns_a_future() 

140 

141 # Triggers execution of async_handler 

142 ee.emit('data', '00101001') 

143 

144 ee.cancel() 

145 

146 # async_handler execution has been canceled 

147 ``` 

148 

149 This is useful if you're attempting to shut down your application and 

150 attempts at a graceful shutdown via `wait_for_complete` have failed. 

151 """ 

152 for fut in self._waiting: 

153 if not fut.done() and not fut.cancelled(): 

154 fut.cancel() 

155 self._waiting.clear() 

156 

157 @property 

158 def complete(self: Self) -> bool: 

159 """When true, there are no pending tasks, and execution is complete. 

160 For example: 

161 

162 ```py 

163 @ee.on('event') 

164 async def async_handler(*args, **kwargs): 

165 await returns_a_future() 

166 

167 # Triggers execution of async_handler 

168 ee.emit('data', '00101001') 

169 

170 # async_handler is still running, so this prints False 

171 print(ee.complete) 

172 

173 await ee.wait_for_complete() 

174 

175 # async_handler has completed execution, so this prints True 

176 print(ee.complete) 

177 ``` 

178 """ 

179 return not self._waiting