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

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

2 

3from collections import OrderedDict 

4from threading import Lock 

5from typing import ( 

6 Any, 

7 Callable, 

8 Dict, 

9 List, 

10 Mapping, 

11 Optional, 

12 overload, 

13 Set, 

14 Tuple, 

15 TypeVar, 

16 Union, 

17) 

18 

19Self = Any 

20 

21 

22class PyeeException(Exception): 

23 """An exception internal to pyee. Deprecated in favor of PyeeError.""" 

24 

25 

26class PyeeError(PyeeException): 

27 """An error internal to pyee.""" 

28 

29 

30Handler = TypeVar("Handler", bound=Callable) 

31 

32 

33class EventEmitter: 

34 """The base event emitter class. All other event emitters inherit from 

35 this class. 

36 

37 Most events are registered with an emitter via the `on` and `once` 

38 methods, and fired with the `emit` method. However, pyee event emitters 

39 have two *special* events: 

40 

41 - `new_listener`: Fires whenever a new listener is created. Listeners for 

42 this event do not fire upon their own creation. 

43 

44 - `error`: When emitted raises an Exception by default, behavior can be 

45 overridden by attaching callback to the event. 

46 

47 For example: 

48 

49 ```py 

50 @ee.on('error') 

51 def on_error(message): 

52 logging.err(message) 

53 

54 ee.emit('error', Exception('something blew up')) 

55 ``` 

56 

57 All callbacks are handled in a synchronous, blocking manner. As in node.js, 

58 raised exceptions are not automatically handled for you---you must catch 

59 your own exceptions, and treat them accordingly. 

60 """ 

61 

62 def __init__(self: Self) -> None: 

63 self._events: Dict[ 

64 str, 

65 "OrderedDict[Callable, Callable]", 

66 ] = dict() 

67 self._lock: Lock = Lock() 

68 

69 def __getstate__(self: Self) -> Mapping[str, Any]: 

70 state = self.__dict__.copy() 

71 del state["_lock"] 

72 return state 

73 

74 def __setstate__(self: Self, state: Mapping[str, Any]) -> None: 

75 self.__dict__.update(state) 

76 self._lock = Lock() 

77 

78 @overload 

79 def on(self: Self, event: str) -> Callable[[Handler], Handler]: ... 

80 @overload 

81 def on(self: Self, event: str, f: Handler) -> Handler: ... 

82 

83 def on( 

84 self: Self, event: str, f: Optional[Handler] = None 

85 ) -> Union[Handler, Callable[[Handler], Handler]]: 

86 """Registers the function `f` to the event name `event`, if provided. 

87 

88 If `f` isn't provided, this method calls `EventEmitter#listens_to`, and 

89 otherwise calls `EventEmitter#add_listener`. In other words, you may either 

90 use it as a decorator: 

91 

92 ```py 

93 @ee.on('data') 

94 def data_handler(data): 

95 print(data) 

96 ``` 

97 

98 Or directly: 

99 

100 ```py 

101 ee.on('data', data_handler) 

102 ``` 

103 

104 In both the decorated and undecorated forms, the event handler is 

105 returned. The upshot of this is that you can call decorated handlers 

106 directly, as well as use them in remove_listener calls. 

107 

108 Note that this method's return type is a union type. If you are using 

109 mypy or pyright, you will probably want to use either 

110 `EventEmitter#listens_to` or `EventEmitter#add_listener`. 

111 """ 

112 if f is None: 

113 return self.listens_to(event) 

114 else: 

115 return self.add_listener(event, f) 

116 

117 def listens_to(self: Self, event: str) -> Callable[[Handler], Handler]: 

118 """Returns a decorator which will register the decorated function to 

119 the event name `event`: 

120 

121 ```py 

122 @ee.listens_to("event") 

123 def data_handler(data): 

124 print(data) 

125 ``` 

126 

127 By only supporting the decorator use case, this method has improved 

128 type safety over `EventEmitter#on`. 

129 """ 

130 

131 def on(f: Handler) -> Handler: 

132 self._add_event_handler(event, f, f) 

133 return f 

134 

135 return on 

136 

137 def add_listener(self: Self, event: str, f: Handler) -> Handler: 

138 """Register the function `f` to the event name `event`: 

139 

140 ``` 

141 def data_handler(data): 

142 print(data) 

143 

144 h = ee.add_listener("event", data_handler) 

145 ``` 

146 

147 By not supporting the decorator use case, this method has improved 

148 type safety over `EventEmitter#on`. 

149 """ 

150 self._add_event_handler(event, f, f) 

151 return f 

152 

153 def _add_event_handler(self: Self, event: str, k: Callable, v: Callable): 

154 # Fire 'new_listener' *before* adding the new listener! 

155 self.emit("new_listener", event, k) 

156 

157 # Add the necessary function 

158 # Note that k and v are the same for `on` handlers, but 

159 # different for `once` handlers, where v is a wrapped version 

160 # of k which removes itself before calling k 

161 with self._lock: 

162 if event not in self._events: 

163 self._events[event] = OrderedDict() 

164 self._events[event][k] = v 

165 

166 def _emit_run( 

167 self: Self, 

168 f: Callable, 

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

170 kwargs: Dict[str, Any], 

171 ) -> None: 

172 f(*args, **kwargs) 

173 

174 def event_names(self: Self) -> Set[str]: 

175 """Get a set of events that this emitter is listening to.""" 

176 return set(self._events.keys()) 

177 

178 def _emit_handle_potential_error(self: Self, event: str, error: Any) -> None: 

179 if event == "error": 

180 if isinstance(error, Exception): 

181 raise error 

182 else: 

183 raise PyeeError(f"Uncaught, unspecified 'error' event: {error}") 

184 

185 def _call_handlers( 

186 self: Self, 

187 event: str, 

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

189 kwargs: Dict[str, Any], 

190 ) -> bool: 

191 handled = False 

192 

193 with self._lock: 

194 funcs = list(self._events.get(event, OrderedDict()).values()) 

195 for f in funcs: 

196 self._emit_run(f, args, kwargs) 

197 handled = True 

198 

199 return handled 

200 

201 def emit( 

202 self: Self, 

203 event: str, 

204 *args: Any, 

205 **kwargs: Any, 

206 ) -> bool: 

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

208 function. Returns `True` if any functions are attached to `event`; 

209 otherwise returns `False`. 

210 

211 Example: 

212 

213 ```py 

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

215 ``` 

216 

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

218 `data('00101001')'`. 

219 """ 

220 handled = self._call_handlers(event, args, kwargs) 

221 

222 if not handled: 

223 self._emit_handle_potential_error(event, args[0] if args else None) 

224 

225 return handled 

226 

227 def once( 

228 self: Self, 

229 event: str, 

230 f: Optional[Callable] = None, 

231 ) -> Callable: 

232 """The same as `ee.on`, except that the listener is automatically 

233 removed after being called. 

234 """ 

235 

236 def _wrapper(f: Callable) -> Callable: 

237 def g( 

238 *args: Any, 

239 **kwargs: Any, 

240 ) -> Any: 

241 with self._lock: 

242 # Check that the event wasn't removed already right 

243 # before the lock 

244 if event in self._events and f in self._events[event]: 

245 self._remove_listener(event, f) 

246 else: 

247 return None 

248 # f may return a coroutine, so we need to return that 

249 # result here so that emit can schedule it 

250 return f(*args, **kwargs) 

251 

252 self._add_event_handler(event, f, g) 

253 return f 

254 

255 if f is None: 

256 return _wrapper 

257 else: 

258 return _wrapper(f) 

259 

260 def _remove_listener(self: Self, event: str, f: Callable) -> None: 

261 """Naked unprotected removal.""" 

262 if event in self._events: 

263 self._events[event].pop(f) 

264 if not self._events[event]: 

265 del self._events[event] 

266 

267 def remove_listener(self: Self, event: str, f: Callable) -> None: 

268 """Removes the function `f` from `event`.""" 

269 with self._lock: 

270 self._remove_listener(event, f) 

271 

272 def remove_all_listeners(self: Self, event: Optional[str] = None) -> None: 

273 """Remove all listeners attached to `event`. 

274 If `event` is `None`, remove all listeners on all events. 

275 """ 

276 with self._lock: 

277 if event is not None: 

278 self._events[event] = OrderedDict() 

279 else: 

280 self._events = dict() 

281 

282 def listeners(self: Self, event: str) -> List[Callable]: 

283 """Returns a list of all listeners registered to the `event`.""" 

284 return list(self._events.get(event, OrderedDict()).keys())