Restore SQLite database by replacing the database file.
(dump_file: Path)
| 192 | |
| 193 | |
| 194 | def _restore_sqlite(dump_file: Path) -> None: |
| 195 | """Restore SQLite database by replacing the database file.""" |
| 196 | logger.info("Restoring SQLite database...") |
| 197 | db_path = Path(settings.DATABASES["default"]["NAME"]) |
| 198 | backup_current = None |
| 199 | |
| 200 | # Backup current database before overwriting |
| 201 | if db_path.exists(): |
| 202 | backup_current = db_path.with_suffix(".db.bak") |
| 203 | shutil.copy2(db_path, backup_current) |
| 204 | logger.info(f"Backed up current database to {backup_current}") |
| 205 | |
| 206 | # Ensure parent directory exists |
| 207 | db_path.parent.mkdir(parents=True, exist_ok=True) |
| 208 | |
| 209 | # The backup file from _dump_sqlite is a complete SQLite database file |
| 210 | # We can simply copy it over the existing database |
| 211 | shutil.copy2(dump_file, db_path) |
| 212 | |
| 213 | # Verify the restore worked by checking if the file is a readable SQLite database |
| 214 | import sqlite3 as _sqlite3 |
| 215 | try: |
| 216 | conn = _sqlite3.connect(str(db_path)) |
| 217 | conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() |
| 218 | conn.close() |
| 219 | except _sqlite3.DatabaseError as exc: |
| 220 | logger.error(f"SQLite verification failed: {exc}") |
| 221 | if backup_current and backup_current.exists(): |
| 222 | shutil.copy2(backup_current, db_path) |
| 223 | logger.info("Restored original database from backup") |
| 224 | raise RuntimeError(f"SQLite restore verification failed: {exc}") from exc |
| 225 | |
| 226 | logger.info("SQLite restore completed successfully") |
| 227 | |
| 228 | |
| 229 | def create_backup() -> Path: |
no test coverage detected