Coverage for seedboxsync/core/taskmanager/manager.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"""Taskmanager Manager module.""" 

8 

9from pathlib import Path 

10from typing import Any 

11from urllib.parse import urlparse 

12from flask import Flask 

13from huey import SqliteHuey 

14 

15 

16class Manager: 

17 """ 

18 Manage the Huey task queue integration with Flask. 

19 

20 The manager creates a Huey instance from the application configuration, 

21 stores it in the Flask extensions registry, and proxies Huey attributes 

22 through the manager instance. 

23 """ 

24 

25 HUEY_DB_NAME = "huey.db" 

26 HUEY_APP_NAME = "seedboxsync" 

27 

28 def __init__(self, app: Flask | None = None) -> None: 

29 """ 

30 Initialize the Huey manager. 

31 

32 Args: 

33 app: Optional Flask application to initialize immediately. 

34 """ 

35 self.app: Flask | None = None 

36 self.__instance: SqliteHuey | None = None 

37 

38 if app is not None: 

39 self.init_app(app) 

40 

41 def init_app(self, app: Flask) -> None: 

42 """ 

43 Initialize the Huey manager for a Flask application. 

44 

45 The created Huey instance is stored in the Flask extensions registry 

46 under the ``huey`` key. 

47 

48 Args: 

49 app: Flask application to initialize. 

50 """ 

51 self.app = app 

52 

53 huey_instance = self.init_huey() 

54 app.extensions["huey"] = huey_instance 

55 self.__instance = huey_instance 

56 

57 def init_huey(self) -> SqliteHuey: 

58 """ 

59 Create and configure the Huey task queue instance. 

60 

61 Returns: 

62 The configured SQLite-backed Huey instance. 

63 """ 

64 return SqliteHuey( 

65 self.HUEY_APP_NAME, 

66 filename=str(self._get_huey_database()), 

67 ) 

68 

69 def __getattr__(self, name: str) -> Any: 

70 """ 

71 Proxy unknown attributes to the underlying Huey instance. 

72 

73 Args: 

74 name: Name of the requested attribute. 

75 

76 Returns: 

77 The attribute exposed by the underlying Huey instance. 

78 

79 Raises: 

80 RuntimeError: If the manager has not been initialized. 

81 AttributeError: If the Huey instance does not expose the requested 

82 attribute. 

83 """ 

84 if self.__instance is None: 

85 raise RuntimeError("The Huey manager has not been initialized with a Flask application.") 

86 

87 return getattr(self.__instance, name) 

88 

89 def _get_huey_database(self) -> Path: 

90 """ 

91 Build the Huey database path from the application database URL. 

92 

93 The Huey database is created in the same directory as the main SQLite 

94 database configured through ``app.config["DATABASE"]``. 

95 

96 For example, the following database URL: 

97 

98 ``sqlite:////home/user/.config/seedboxsync/seedboxsync.db`` 

99 

100 produces: 

101 

102 ``/home/user/.config/seedboxsync/huey.db`` 

103 

104 Returns: 

105 The path to the Huey SQLite database file. 

106 

107 Raises: 

108 RuntimeError: If the manager has not been initialized with a Flask 

109 application. 

110 ValueError: If the configured database is not a file-based SQLite 

111 database. 

112 """ 

113 if self.app is None: 

114 raise RuntimeError("The Huey manager has not been initialized with a Flask application.") 

115 

116 database_url: str = self.app.config["DATABASE"] 

117 parsed_url = urlparse(database_url) 

118 

119 clean_path = parsed_url.path 

120 if clean_path.startswith("//"): 

121 clean_path = clean_path[1:] 

122 

123 if parsed_url.scheme != "sqlite": 

124 raise ValueError(f"Unsupported database scheme: {parsed_url.scheme or '<missing>'}") 

125 

126 if not parsed_url.path: 

127 raise ValueError("The SQLite database URL does not contain a file path.") 

128 

129 if parsed_url.path in {":memory:", "/:memory:"}: 

130 raise ValueError("An in-memory SQLite database cannot be used to derive the Huey database path.") 

131 

132 database_path = Path(clean_path) 

133 huey_database_path = database_path.with_name(self.HUEY_DB_NAME) 

134 

135 self.app.logger.debug( 

136 "Using Huey task queue database path: %s", 

137 huey_database_path, 

138 ) 

139 

140 return huey_database_path