Coverage for seedboxsync/core/taskmanager/manager.py: 98%

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 

9import os 

10from pathlib import Path 

11import re 

12from typing import Any 

13from urllib.parse import unquote, urlparse 

14from flask import Flask 

15from huey import SqliteHuey 

16 

17 

18class Manager: 

19 """ 

20 Manage the Huey task queue integration with Flask. 

21 

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

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

24 through the manager instance. 

25 """ 

26 

27 HUEY_DB_NAME = "huey.db" 

28 HUEY_APP_NAME = "seedboxsync" 

29 

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

31 """ 

32 Initialize the Huey manager. 

33 

34 Args: 

35 app: Optional Flask application to initialize immediately. 

36 """ 

37 self.app: Flask | None = None 

38 self.__instance: SqliteHuey | None = None 

39 

40 if app is not None: 

41 self.init_app(app) 

42 

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

44 """ 

45 Initialize the Huey manager for a Flask application. 

46 

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

48 under the ``huey`` key. 

49 

50 Args: 

51 app: Flask application to initialize. 

52 """ 

53 self.app = app 

54 

55 huey_instance = self.init_huey() 

56 app.extensions["huey"] = huey_instance 

57 self.__instance = huey_instance 

58 

59 def init_huey(self) -> SqliteHuey: 

60 """ 

61 Create and configure the Huey task queue instance. 

62 

63 Returns: 

64 The configured SQLite-backed Huey instance. 

65 """ 

66 return SqliteHuey( 

67 self.HUEY_APP_NAME, 

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

69 ) 

70 

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

72 """ 

73 Proxy unknown attributes to the underlying Huey instance. 

74 

75 Args: 

76 name: Name of the requested attribute. 

77 

78 Returns: 

79 The attribute exposed by the underlying Huey instance. 

80 

81 Raises: 

82 RuntimeError: If the manager has not been initialized. 

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

84 attribute. 

85 """ 

86 if self.__instance is None: 

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

88 

89 return getattr(self.__instance, name) 

90 

91 def _get_huey_database(self) -> Path: 

92 """ 

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

94 

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

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

97 

98 For example, the following database URL: 

99 

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

101 

102 produces: 

103 

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

105 

106 Returns: 

107 The path to the Huey SQLite database file. 

108 

109 Raises: 

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

111 application. 

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

113 database. 

114 """ 

115 if self.app is None: 

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

117 

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

119 parsed_url = urlparse(database_url) 

120 

121 clean_path = unquote(parsed_url.path) 

122 

123 # SQLite URI represents Windows drive paths as /C:/... 

124 if os.name == "nt" and re.match(r"^/[A-Za-z]:/", clean_path): 

125 clean_path = clean_path[1:] 

126 

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

128 clean_path = clean_path[1:] 

129 

130 if parsed_url.scheme != "sqlite": 

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

132 

133 if not parsed_url.path: 

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

135 

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

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

138 

139 database_path = Path(clean_path) 

140 huey_database_path = database_path.with_name(self.HUEY_DB_NAME) 

141 

142 self.app.logger.debug( 

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

144 huey_database_path, 

145 ) 

146 

147 return huey_database_path