(self)
| 67 | query_options = {"sort": ["type"]} |
| 68 | |
| 69 | def __init__(self): |
| 70 | self._timers = TimersHolder() |
| 71 | self._trigger_types = TIMER_TRIGGER_TYPES.keys() |
| 72 | queue_suffix = self.__class__.__name__ |
| 73 | self._trigger_watcher = TriggerWatcher( |
| 74 | create_handler=self._handle_create_trigger, |
| 75 | update_handler=self._handle_update_trigger, |
| 76 | delete_handler=self._handle_delete_trigger, |
| 77 | trigger_types=self._trigger_types, |
| 78 | queue_suffix=queue_suffix, |
| 79 | exclusive=True, |
| 80 | ) |
| 81 | self._trigger_watcher.start() |
| 82 | self._register_timer_trigger_types() |
| 83 | self._allowed_timer_types = TIMER_TRIGGER_TYPES.keys() |
| 84 | |
| 85 | def get_all(self, timer_type=None): |
| 86 | if timer_type and timer_type not in self._allowed_timer_types: |
nothing calls this directly
no test coverage detected