Coverage for seedboxsync/core/db.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"""Database module.""" 

8 

9import os 

10from os import fspath 

11from pathlib import Path 

12from typing import ClassVar, cast 

13from flask import Flask 

14from humanize import filesize, time 

15from peewee import SqliteDatabase 

16from playhouse.flask_utils import FlaskDB 

17from playhouse.migrate import SchemaMigrator, migrate 

18from seedboxsync.core import utils 

19from seedboxsync.core.dao import Download, SeedboxSync, TaskStatus, Torrent 

20 

21 

22class Database: 

23 """ 

24 Database connector using peewee. 

25 

26 Attributes: 

27 app (Flask): The Flask application that owns the database connection. 

28 """ 

29 

30 DATABASE_VERSION = 4 

31 DB_PATHS: ClassVar[list[Path]] = [ 

32 Path("~/.config/seedboxsync/seedboxsync.db").expanduser().resolve(), 

33 Path("~/.seedboxsync.db").expanduser().resolve(), 

34 Path("~/.seedboxsync/config/seedboxsync.db").expanduser().resolve(), 

35 Path("/etc/seedboxsync/seedboxsync.db"), 

36 ] 

37 db: SqliteDatabase 

38 

39 def __init__(self, app: Flask) -> None: 

40 """ 

41 Initialize a new Database instance. 

42 

43 Args: 

44 app (Flask): The Flask application to bind to the database. 

45 """ 

46 self.app = app 

47 self._load_database() 

48 self._register_functions() 

49 

50 def _load_database(self) -> None: 

51 """Load SeedboxSync DB from SeedboxSyncFront.""" 

52 if self.app.config.get("DATABASE", False): 

53 # Load from testing 

54 self._db_file = self.app.config.get("DATABASE", "") 

55 self.app.config["DATABASE"] = "sqlite:///" + self._db_file 

56 else: 

57 # Get DB from paths, default to first path if none found 

58 self.app.config.setdefault("DATABASE", fspath(Database.DB_PATHS[0])) # default path 

59 for path in Database.DB_PATHS: 

60 if path.exists() and path.is_file() and os.access(path, os.W_OK): 

61 self.app.config.setdefault("DATABASE", fspath(path)) 

62 self.app.logger.debug("Use database path %s", path) 

63 self._db_file = self.app.config["DATABASE"] 

64 self.app.config["DATABASE"] = "sqlite:///" + self._db_file 

65 

66 if not Path(self._db_file).exists(): 

67 self.app.logger.warning(f'Database "{self._db_file}" not found — creating new file...') 

68 utils.ensure_dir_exists(Path(self._db_file).parent) 

69 self._init_and_bind() 

70 self._create_db_schema() 

71 else: 

72 self._init_and_bind() 

73 

74 # Check and run migrations if needed 

75 db_version = int(SeedboxSync.get_db_version()) 

76 self.app.logger.debug(f"SQLite database version is {db_version}") 

77 while db_version < self.DATABASE_VERSION: 

78 next_version = db_version + 1 

79 migration_name = f"migrate_to_{next_version}" 

80 

81 self.app.logger.info(f'Upgrading database "{self._db_file}" from v{db_version} to v{next_version}') 

82 

83 # Dynamically resolve migration function 

84 migration_func = getattr(self, migration_name, None) 

85 if migration_func is None: 

86 raise RuntimeError(f"Missing migration function: {migration_name}") 

87 migration_func() 

88 db_version = next_version 

89 

90 def _init_and_bind(self) -> None: 

91 """Initialize and bind Peewee models to the SQLite database.""" 

92 db_wrapper = FlaskDB(self.app) 

93 self.db = cast(SqliteDatabase, db_wrapper.database) 

94 self.app.extensions["flaskdb"] = db_wrapper 

95 self.db.journal_mode = "wal" 

96 self.db.cache_size = -64000 

97 self.db.foreign_keys = 1 

98 self.db.bind([Download, SeedboxSync, TaskStatus, Torrent]) 

99 self.app.logger.debug( 

100 "Database initialized %s / journal_mode=%s, cache_size=%s, foreign_keys=%s", 

101 self.app.config["DATABASE"], 

102 self.db.journal_mode, 

103 self.db.cache_size, 

104 self.db.foreign_keys, 

105 ) 

106 

107 def _register_functions(self) -> None: 

108 """Register DB functions.""" 

109 

110 @self.db.func("byte_to_gi") 

111 def db_byte_to_gi(num: float, suffix: str = "B") -> str: # pyright: ignore [reportUnusedFunction] 

112 return utils.byte_to_gi(num, suffix) 

113 

114 @self.db.func("humanize") 

115 def db_humanize(num: float) -> str: # pyright: ignore [reportUnusedFunction] 

116 try: 

117 # Treat None or invalid type as 0 

118 num = float(num or 0) 

119 except (ValueError, TypeError): 

120 num = 0.0 

121 return filesize.naturalsize(num, True) 

122 

123 @self.db.func("naturaldelta") 

124 def db_naturaldelta(num: float) -> str: # pyright: ignore [reportUnusedFunction] 

125 try: 

126 # Treat None or invalid type as 0 

127 num = float(num or 0) 

128 except (ValueError, TypeError): 

129 num = 0.0 

130 return time.naturaldelta(num, minimum_unit="seconds", months=False) 

131 

132 # 

133 # Database creation and migration 

134 # 

135 def _create_db_schema(self) -> None: 

136 """Create all tables and set the initial database version.""" 

137 self.db.create_tables([Download, Torrent, TaskStatus, SeedboxSync]) 

138 SeedboxSync.set_db_version(str(self.DATABASE_VERSION)) 

139 

140 def migrate_to_2(self) -> None: 

141 """ 

142 Migration: rebuild SeedboxSync table and add Lock table. 

143 

144 Fixes compatibility issues between tables created with Peewee v2 and v3. 

145 """ 

146 self.db.drop_tables([SeedboxSync]) 

147 self.db.create_tables([SeedboxSync]) 

148 SeedboxSync.set_db_version("2") 

149 

150 def migrate_to_3(self) -> None: 

151 """Migration: allow null values for the 'announce' field in the torrent table.""" 

152 migrator = SchemaMigrator.from_database(self.db) 

153 migrate( 

154 migrator.drop_not_null("torrent", "announce"), 

155 ) 

156 SeedboxSync.set_db_version("3") 

157 

158 def migrate_to_4(self) -> None: 

159 """Replace 'Lock' table by 'TaskStatus'.""" 

160 self.db.execute_sql("DROP TABLE IF EXISTS lock;") # type: ignore[no-untyped-call] 

161 self.db.execute_sql("DELETE FROM seedboxsync WHERE key = 'version';") # type: ignore[no-untyped-call] 

162 self.db.create_tables([TaskStatus]) 

163 

164 SeedboxSync.set_db_version("4")