Coverage for /home/ubuntu/shekels/python/shekels/server/event_listener.py: 100%

1from typing import Any, Callable # noqa: F401 

2 

3from collections import deque 

4from copy import deepcopy 

5from lunchbox.enforce import Enforce 

6import dash 

7# ------------------------------------------------------------------------------ 

8 

9 

10class EventListener: 

11 ''' 

12 Listens for Dash app events and calls registered callbacks. 

13 ''' 

14 def __init__(self, app, store, memory=10): 

15 # type: (dash.Dash, dict, int) -> None 

16 ''' 

17 Constructs EventListener. 

18 

19 Args: 

20 app (dash.Dash): Dash application instance. 

21 store (dict): Dash store. 

22 memory (int, optional): Number of state changes to remember. 

23 Default: 10. 

24 

25 Raises: 

26 EnforceError: If app is not an instance of dash.Dash. 

27 EnforceError: If app is not an instance of dict. 

28 EnforceError: If memory is less than 1. 

29 ''' 

30 Enforce(app, 'instance of', dash.Dash) 

31 Enforce(store, 'instance of', dict) 

32 msg = 'Memory must be greater or equal to {b}. {a} < {b}.' 

33 Enforce(memory, '>=', 1, message=msg) 

34 # ---------------------------------------------------------------------- 

35 

36 self._app = app # type: dash.Dash 

37 self.events = {} # type: dict 

38 self.state = deque([deepcopy(store)], memory) # type: deque 

39 

40 def listen(self, event, callback): 

41 # type: (str, Callable[[Any, dict, dash.Dash], dict]) -> EventListener 

42 ''' 

43 Listen for given event and call given callback. 

44 

45 Args: 

46 event (str): Event name. 

47 callback (function): Function of form (value, store, app) -> store. 

48 

49 Raises: 

50 EnforceError: If event is not a string. 

51 

52 Returns: 

53 EventListener: self. 

54 ''' 

55 msg = 'Event name must be a string. {a} is not a string.' 

56 Enforce(event, 'instance of', str, message=msg) 

57 self.events[event] = callback 

58 return self 

59 

60 def emit(self, event, value): 

61 # type: (str, object) -> EventListener 

62 ''' 

63 Call a registered callback guven an event and value. 

64 

65 Args: 

66 event (str): Event name. 

67 value (object): Value to be given to callback. 

68 

69 Raises: 

70 EnforceError: If event is not a string. 

71 

72 Returns: 

73 EventListener: self. 

74 ''' 

75 msg = 'Event name must be a string. {a} is not a string.' 

76 Enforce(event, 'instance of', str, message=msg) 

77 if event in self.events: 

78 store = self.events[event](value, self.store, self._app) 

79 self.state.append(deepcopy(store)) 

80 return self 

81 

82 @property 

83 def store(self): 

84 # type () -> dict 

85 ''' 

86 dict: Copy of last item in state. 

87 ''' 

88 return deepcopy(self.state[-1])