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
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 03:04 +0000
1from collections.abc import Callable, Iterable
3from mealie.core import root_logger
5logger = root_logger.get_logger()
8class SchedulerRegistry:
9 """
10 A container class for registering and removing callbacks for the scheduler.
11 """
13 _daily: list[Callable] = []
14 _hourly: list[Callable] = []
15 _minutely: list[Callable] = []
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)
23 @staticmethod
24 def register_daily(*callbacks: Callable):
25 SchedulerRegistry._register("daily", SchedulerRegistry._daily, callbacks)
27 @staticmethod
28 def remove_daily(callback: Callable):
29 logger.debug(f"Removing daily callback: {callback.__name__}")
30 SchedulerRegistry._daily.remove(callback)
32 @staticmethod
33 def register_hourly(*callbacks: Callable):
34 SchedulerRegistry._register("daily", SchedulerRegistry._hourly, callbacks)
36 @staticmethod
37 def remove_hourly(callback: Callable):
38 logger.debug(f"Removing hourly callback: {callback.__name__}")
39 SchedulerRegistry._hourly.remove(callback)
41 @staticmethod
42 def register_minutely(*callbacks: Callable):
43 SchedulerRegistry._register("minutely", SchedulerRegistry._minutely, callbacks)
45 @staticmethod
46 def remove_minutely(callback: Callable):
47 logger.debug(f"Removing minutely callback: {callback.__name__}")
48 SchedulerRegistry._minutely.remove(callback)
50 @staticmethod
51 def print_jobs():
52 for job in SchedulerRegistry._daily:
53 logger.debug(f"Daily job: {job.__name__}")
55 for job in SchedulerRegistry._hourly:
56 logger.debug(f"Hourly job: {job.__name__}")
58 for job in SchedulerRegistry._minutely:
59 logger.debug(f"Minutely job: {job.__name__}")