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

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, Log, SeedboxSync, TaskStatus, Torrent, User 

16 

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

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

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

20 

21 

22class DatabaseMigration: 

23 """Manage database schema migrations.""" 

24 

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

26 """ 

27 Initialize the database migration manager. 

28 

29 Args: 

30 app: Flask application instance. 

31 db: Peewee SQLite database instance. 

32 """ 

33 self.app = app 

34 self.db = db 

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

36 

37 def upgrade(self) -> None: 

38 """ 

39 Upgrade the database schema to the latest available migration. 

40 

41 Existing databases using the legacy integer-based migration system 

42 are first converted to the Peewee migration history format. 

43 """ 

44 with self.app.app_context(): 

45 self._run_migrations() 

46 self._set_last_migration() 

47 

48 def _run_migrations(self) -> None: 

49 """Apply all pending Peewee migrations.""" 

50 applied = self.runner.up() 

51 

52 for migration in applied: 

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

54 

55 def _set_last_migration(self) -> None: 

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

57 migrations = self.runner.status() 

58 if migrations: 

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