Coverage for seedboxsync/core/taskmanager/utils.py: 100%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""Task queue initialization and automatic task module registration.""" 

8 

9import importlib 

10import pkgutil 

11from types import ModuleType 

12from seedboxsync.core import current_app as app 

13import seedboxsync.core.taskmanager.task as task_package 

14 

15 

16def load_task_modules() -> list[ModuleType]: 

17 """ 

18 Import all task modules from this package. 

19 

20 Modules are loaded when their name starts with ``task_``. Importing them 

21 registers their decorated Huey tasks in the task queue registry. 

22 

23 Returns: 

24 The list of imported task modules. 

25 """ 

26 imported_modules: list[ModuleType] = [] 

27 

28 for module_info in pkgutil.iter_modules(task_package.__path__): 

29 if not module_info.name.startswith("task_"): 

30 continue 

31 

32 module_name = f"{task_package.__name__}.{module_info.name}" 

33 app.logger.debug("Register task on %s", module_name) 

34 

35 module = importlib.import_module(module_name) 

36 imported_modules.append(module) 

37 

38 return imported_modules