Coverage for seedboxsync/core/database/database.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 

9from datetime import datetime 

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 seedboxsync.core import utils 

18from seedboxsync.core.database.migration import DatabaseMigration 

19from seedboxsync.core.database.models import ApiKey, Download, SeedboxSync, TaskStatus, Torrent, User 

20 

21 

22class Database: 

23 """ 

24 Database manager for SeedboxSync using Peewee ORM. 

25 

26 Handles database path resolution, connection binding, SQLite optimization 

27 pragmas, schema migrations, and custom SQLite functions. 

28 

29 Attributes: 

30 DB_PATHS (ClassVar[list[Path]]): Candidate database paths checked in order of preference. 

31 app (Flask): The Flask application instance bound to this database. 

32 db (SqliteDatabase): The initialized Peewee SQLite database instance. 

33 """ 

34 

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

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

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

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

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

40 ] 

41 db: SqliteDatabase 

42 

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

44 """ 

45 Initialize a new Database instance. 

46 

47 Args: 

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

49 """ 

50 self.app = app 

51 self._load_database() 

52 self._register_functions() 

53 

54 def _load_database(self) -> None: 

55 """ 

56 Locate, configure, and initialize the SQLite database. 

57 

58 Checks candidate file paths or pre-configured settings, creates parent 

59 directories if necessary, builds the schema for new files, and executes 

60 pending migrations. 

61 

62 Raises: 

63 RuntimeError: If a required migration method is missing. 

64 """ 

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

66 # Load from testing 

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

68 self.app.config["DATABASE"] = f"sqlite:///{Path(self._db_file).as_posix()}" 

69 else: 

70 # Get DB from path 

71 db_path = utils.get_database_path_from_paths() 

72 self.app.config["DATABASE"] = fspath(db_path) 

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

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

75 self.app.config["DATABASE"] = f"sqlite:///{Path(self._db_file).as_posix()}" 

76 

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

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

79 self.app.config["MIGRATE_DB"] = True 

80 

81 self._init_and_bind() 

82 

83 def _init_and_bind(self) -> None: 

84 """ 

85 Initialize the Flask-Peewee wrapper and apply performance tuning pragmas. 

86 

87 Binds model classes to the database connection and configures WAL mode, 

88 cache size, and foreign key constraints. 

89 """ 

90 db_wrapper = FlaskDB(self.app) 

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

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

93 self.db.journal_mode = "wal" 

94 self.db.cache_size = -64000 

95 self.db.foreign_keys = 1 

96 self.db.bind([ApiKey, Download, SeedboxSync, TaskStatus, Torrent, User]) 

97 self.app.logger.debug( 

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

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

100 self.db.journal_mode, 

101 self.db.cache_size, 

102 self.db.foreign_keys, 

103 ) 

104 if self.app.config.get("MIGRATE_DB", False): 

105 DatabaseMigration(self.app, self.db).upgrade() 

106 

107 def _register_functions(self) -> None: 

108 """Register custom SQLite scalar functions for SQL queries.""" 

109 

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

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

112 """Convert byte counts to human-readable binary unit strings.""" 

113 return utils.byte_to_gi(num, suffix) 

114 

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

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

117 """Format file size numbers into human-readable representations.""" 

118 if num is None: 

119 return "" 

120 try: 

121 # Treat None or invalid type as 0 

122 num = float(num or 0) 

123 except (ValueError, TypeError): 

124 return "" 

125 return filesize.naturalsize(num, True) 

126 

127 @self.db.func("short_datetime") 

128 def db_short_datetime(value: str | None) -> str | None: # pyright: ignore[reportUnusedFunction] 

129 """Format a datetime without microseconds.""" 

130 if value is None: 

131 return None 

132 return datetime.fromisoformat(value).strftime("%Y-%m-%d %H:%M:%S") 

133 

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

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

136 """Format second intervals into human-readable duration strings.""" 

137 try: 

138 # Treat None or invalid type as 0 

139 num = float(num or 0) 

140 except (ValueError, TypeError): 

141 num = 0.0 

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