diff --git a/apps/backups/services.py b/apps/backups/services.py index c70160e1..98f563cf 100644 --- a/apps/backups/services.py +++ b/apps/backups/services.py @@ -171,30 +171,25 @@ def _restore_postgresql(dump_file: Path) -> None: def _dump_sqlite(output_file: Path) -> None: - """Dump SQLite database using sqlite3 .backup command.""" - logger.info("Dumping SQLite database with sqlite3 .backup...") + import sqlite3 as _sqlite3 + logger.info("Dumping SQLite database...") db_path = Path(settings.DATABASES["default"]["NAME"]) if not db_path.exists(): raise FileNotFoundError(f"SQLite database not found: {db_path}") - # Use sqlite3 .backup command via stdin for reliable execution - result = subprocess.run( - ["sqlite3", str(db_path)], - input=f".backup '{output_file}'\n", - capture_output=True, - text=True, - ) + src = _sqlite3.connect(str(db_path)) + dst = _sqlite3.connect(str(output_file)) + try: + src.backup(dst) + finally: + dst.close() + src.close() - if result.returncode != 0: - logger.error(f"sqlite3 backup failed: {result.stderr}") - raise RuntimeError(f"sqlite3 backup failed: {result.stderr}") - - # Verify the backup file was created if not output_file.exists(): - raise RuntimeError("sqlite3 backup failed: output file not created") + raise RuntimeError("SQLite backup failed: output file not created") - logger.info(f"sqlite3 backup completed successfully: {output_file}") + logger.info(f"SQLite backup completed successfully: {output_file}") def _restore_sqlite(dump_file: Path) -> None: @@ -216,23 +211,20 @@ def _restore_sqlite(dump_file: Path) -> None: # We can simply copy it over the existing database shutil.copy2(dump_file, db_path) - # Verify the restore worked by checking if sqlite3 can read it - result = subprocess.run( - ["sqlite3", str(db_path)], - input=".tables\n", - capture_output=True, - text=True, - ) - - if result.returncode != 0: - logger.error(f"sqlite3 verification failed: {result.stderr}") - # Try to restore from backup + # Verify the restore worked by checking if the file is a readable SQLite database + import sqlite3 as _sqlite3 + try: + conn = _sqlite3.connect(str(db_path)) + conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() + conn.close() + except _sqlite3.DatabaseError as exc: + logger.error(f"SQLite verification failed: {exc}") if backup_current and backup_current.exists(): shutil.copy2(backup_current, db_path) logger.info("Restored original database from backup") - raise RuntimeError(f"sqlite3 restore verification failed: {result.stderr}") + raise RuntimeError(f"SQLite restore verification failed: {exc}") from exc - logger.info("sqlite3 restore completed successfully") + logger.info("SQLite restore completed successfully") def create_backup() -> Path: