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

41 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-07 03:04 +0000

1from collections.abc import Callable, Iterable 

2 

3from mealie.core import root_logger 

4 

5logger = root_logger.get_logger() 

6 

7 

8class SchedulerRegistry: 

9 """ 

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

11 """ 

12 

13 _daily: list[Callable] = [] 

14 _hourly: list[Callable] = [] 

15 _minutely: list[Callable] = [] 

16 

17 @staticmethod 

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

19 for cb in callback: 

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

21 callbacks.append(cb) 

22 

23 @staticmethod 

24 def register_daily(*callbacks: Callable): 

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

26 

27 @staticmethod 

28 def remove_daily(callback: Callable): 

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

30 SchedulerRegistry._daily.remove(callback) 

31 

32 @staticmethod 

33 def register_hourly(*callbacks: Callable): 

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

35 

36 @staticmethod 

37 def remove_hourly(callback: Callable): 

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

39 SchedulerRegistry._hourly.remove(callback) 

40 

41 @staticmethod 

42 def register_minutely(*callbacks: Callable): 

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

44 

45 @staticmethod 

46 def remove_minutely(callback: Callable): 

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

48 SchedulerRegistry._minutely.remove(callback) 

49 

50 @staticmethod 

51 def print_jobs(): 

52 for job in SchedulerRegistry._daily: 

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

54 

55 for job in SchedulerRegistry._hourly: 

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

57 

58 for job in SchedulerRegistry._minutely: 

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