Coverage for seedboxsync/core/database/migration.py: 67%

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 

8"""Database schema migration management.""" 

9 

10from pathlib import Path 

11from flask import Flask 

12from peewee import SqliteDatabase 

13from playhouse.migrations import Runner 

14from seedboxsync.core import utils 

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

16 

17database = SqliteDatabase(str(utils.get_database_path_from_paths())) 

18models = [ApiKey, Download, SeedboxSync, TaskStatus, Torrent, User] 

19MIGRATION_PATH = str(Path(__file__).parent / "migrations") 

20 

21# Map each legacy database version to the latest equivalent migration 

22# managed by the Peewee migration runner. 

23LEGACY_MIGRATION_MAP: dict[int, str | None] = { 

24 1: "0001_initial", 

25 2: None, 

26 3: "0002_torrent_nullable_announce", 

27 4: "0003_add_taskstatus", 

28} 

29 

30 

31class DatabaseMigration: 

32 """Manage database schema migrations.""" 

33 

34 def __init__(self, app: Flask, db: SqliteDatabase) -> None: 

35 """ 

36 Initialize the database migration manager. 

37 

38 Args: 

39 app: Flask application instance. 

40 db: Peewee SQLite database instance. 

41 """ 

42 self.app = app 

43 self.db = db 

44 self.runner = Runner(self.db, directory=MIGRATION_PATH) 

45 

46 def upgrade(self) -> None: 

47 """ 

48 Upgrade the database schema to the latest available migration. 

49 

50 Existing databases using the legacy integer-based migration system 

51 are first converted to the Peewee migration history format. 

52 """ 

53 self._migrate_legacy_database() 

54 with self.app.app_context(): 

55 self._run_migrations() 

56 self._set_last_migration() 

57 

58 def _run_migrations(self) -> None: 

59 """Apply all pending Peewee migrations.""" 

60 applied = self.runner.up() 

61 

62 for migration in applied: 

63 self.app.logger.info("Applied database migration: %s", migration) 

64 

65 def _set_last_migration(self) -> None: 

66 """Store the latest available migration name in the Flask configuration.""" 

67 migrations = self.runner.status() 

68 if migrations: 

69 self.app.config["LAST_MIGRATION"] = migrations[-1].name 

70 

71 def _migrate_legacy_database(self) -> None: 

72 """ 

73 Convert a database using the legacy migration system. 

74 

75 Legacy SeedboxSync releases stored an integer database version in the 

76 ``seedboxsync`` table. The corresponding Peewee migrations are marked 

77 as already applied without executing them again. 

78 

79 Databases that do not contain a legacy version are left untouched. 

80 """ 

81 db_version = self._get_legacy_database_version() 

82 

83 if db_version is None: 

84 return 

85 

86 self.app.logger.info("Legacy database detected at version %d; converting migration history to Peewee migrations.", db_version) 

87 for version, migration_name in LEGACY_MIGRATION_MAP.items(): 

88 if version > db_version: 

89 break 

90 if migration_name is None: 

91 continue 

92 self.app.logger.info("Marking legacy database version %d as migration '%s' already applied.", version, migration_name) 

93 self.runner.fake(migration_name) 

94 

95 self._remove_legacy_database_version() 

96 

97 def _get_legacy_database_version(self) -> int | None: 

98 """ 

99 Return the legacy database version if one exists. 

100 

101 Returns: 

102 The legacy database version, or ``None`` when the database does 

103 not use the legacy migration system. 

104 """ 

105 if "seedboxsync" not in self.db.get_tables(): 

106 return None 

107 

108 version = SeedboxSync.select(SeedboxSync.value).where(SeedboxSync.key == "db_version").first() 

109 

110 if version is None or not version.value: 

111 return None 

112 

113 try: 

114 return int(version.value) 

115 except (TypeError, ValueError) as exc: 

116 raise RuntimeError(f"Invalid legacy database version: {version.value!r}") from exc 

117 

118 def _remove_legacy_database_version(self) -> None: 

119 """Remove the obsolete legacy database version entry.""" 

120 (SeedboxSync.delete().where(SeedboxSync.key == "db_version").execute()) 

121 self.app.logger.debug("Removed legacy database version metadata.")