Coverage for seedboxsync/taskmanager.py: 97%

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"""Starter for the SeedboxSync taskmanager.""" 

8 

9import logging 

10import os 

11from typing import Any 

12from seedboxsync import create_app 

13from seedboxsync.core import Config 

14from seedboxsync.core.taskmanager import heartbeat_shutdown, heartbeat_startup 

15from seedboxsync.core.taskmanager.utils import load_task_modules 

16 

17log_level = logging.getLevelNamesMapping().get(os.getenv("HUEY_LOG_LEVEL", "INFO").upper(), logging.INFO) 

18app = create_app() 

19 

20with app.app_context(): 

21 huey = app.task_manager 

22 app.logger.setLevel(log_level) 

23 app.logger.info("Start huey consumer") 

24 

25 @huey.on_startup() # type: ignore[untyped-decorator] 

26 def on_startup() -> None: 

27 """Call function on huey startup.""" 

28 heartbeat_startup() 

29 app.logger.debug("Flushing old tasks from queue...") 

30 huey.flush() 

31 

32 @huey.on_shutdown() # type: ignore[untyped-decorator] 

33 def on_shutdown() -> None: 

34 """Call function on huey shutdown.""" 

35 heartbeat_shutdown() 

36 

37 @huey.pre_execute() # type: ignore[untyped-decorator] 

38 def setup_worker_logging(task: Any) -> None: 

39 """Configure the Huey logger from Flask.""" 

40 huey_logger = logging.getLogger("huey") 

41 huey_logger.handlers = [] 

42 for handler in app.logger.handlers: 

43 huey_logger.addHandler(handler) 

44 

45 @huey.pre_execute() # type: ignore[untyped-decorator] 

46 def reload_config(task: Any) -> None: 

47 """Reload Flask DB config.""" 

48 config = Config.reload_config(app) 

49 app.config.from_mapping(config) 

50 

51 load_task_modules()