Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/scheduler/scheduler_registry.py: 88%

41 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:48 +0000

1from collections.abc import Callable, Iterable 1a

2 

3from mealie.core import root_logger 1a

4 

5logger = root_logger.get_logger() 1a

6 

7 

8class SchedulerRegistry: 1a

9 """ 

10 A container class for registering and removing callbacks for the scheduler. 

11 """ 

12 

13 _daily: list[Callable] = [] 1a

14 _hourly: list[Callable] = [] 1a

15 _minutely: list[Callable] = [] 1a

16 

17 @staticmethod 1a

18 def _register(name: str, callbacks: list[Callable], callback: Iterable[Callable]): 1a

19 for cb in callback: 1a

20 logger.debug(f"Registering {name} callback: {cb.__name__}") 1a

21 callbacks.append(cb) 1a

22 

23 @staticmethod 1a

24 def register_daily(*callbacks: Callable): 1a

25 SchedulerRegistry._register("daily", SchedulerRegistry._daily, callbacks) 1a

26 

27 @staticmethod 1a

28 def remove_daily(callback: Callable): 1a

29 logger.debug(f"Removing daily callback: {callback.__name__}") 

30 SchedulerRegistry._daily.remove(callback) 

31 

32 @staticmethod 1a

33 def register_hourly(*callbacks: Callable): 1a

34 SchedulerRegistry._register("daily", SchedulerRegistry._hourly, callbacks) 1a

35 

36 @staticmethod 1a

37 def remove_hourly(callback: Callable): 1a

38 logger.debug(f"Removing hourly callback: {callback.__name__}") 

39 SchedulerRegistry._hourly.remove(callback) 

40 

41 @staticmethod 1a

42 def register_minutely(*callbacks: Callable): 1a

43 SchedulerRegistry._register("minutely", SchedulerRegistry._minutely, callbacks) 1a

44 

45 @staticmethod 1a

46 def remove_minutely(callback: Callable): 1a

47 logger.debug(f"Removing minutely callback: {callback.__name__}") 

48 SchedulerRegistry._minutely.remove(callback) 

49 

50 @staticmethod 1a

51 def print_jobs(): 1a

52 for job in SchedulerRegistry._daily: 1a

53 logger.debug(f"Daily job: {job.__name__}") 1a

54 

55 for job in SchedulerRegistry._hourly: 1a

56 logger.debug(f"Hourly job: {job.__name__}") 1a

57 

58 for job in SchedulerRegistry._minutely: 1a

59 logger.debug(f"Minutely job: {job.__name__}") 1a