Mirrored_Repos/InvokeAI
1
0
Fork 0
mirror of https://github.com/invoke-ai/InvokeAI synced 2024-08-30 20:32:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(db): tidy db naming utils

Browse Source
This commit is contained in:
psychedelicious 2023-12-11 01:19:12 +11:00
parent 56966d6d05
commit fcb9e89bd7
1 changed files with 9 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
invokeai/app/services/shared/sqlite/sqlite_migrator.py
View File

@ -117,10 +117,16 @@ class MigrationSet:
def get_temp_db_path(original_db_path: Path) -> Path: def get_temp_db_path(original_db_path: Path) -> Path:
"""Gets the path to the migrated database.""" """Gets the path to the temp database."""
return original_db_path.parent / original_db_path.name.replace(".db", ".db.temp") return original_db_path.parent / original_db_path.name.replace(".db", ".db.temp")
def get_backup_db_path(original_db_path: Path) -> Path:
"""Gets the path to the final backup database."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return original_db_path.parent / f"{original_db_path.stem}_backup_{timestamp}.db"
class SQLiteMigrator: class SQLiteMigrator:
""" """
Manages migrations for a SQLite database. Manages migrations for a SQLite database.
@ -297,11 +303,10 @@ class SQLiteMigrator:
return temp_db_path return temp_db_path
def _finalize_migration(self, temp_db_conn: sqlite3.Connection, temp_db_path: Path, original_db_path: Path) -> None: def _finalize_migration(self, temp_db_conn: sqlite3.Connection, temp_db_path: Path, original_db_path: Path) -> None:
"""Closes connections, renames the original database as a backup and renames the migrated database to the original db path.""" """Closes connections, renames the original database as a backup and renames the migrated database to the original name."""
self._conn.close() self._conn.close()
temp_db_conn.close() temp_db_conn.close()
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_db_path = get_backup_db_path(original_db_path)
backup_db_path = original_db_path.parent / f"{original_db_path.stem}_backup_{timestamp}.db"
original_db_path.rename(backup_db_path) original_db_path.rename(backup_db_path)
temp_db_path.rename(original_db_path) temp_db_path.rename(original_db_path)
self._logger.info(f"Migration successful. Original DB backed up to {backup_db_path}") self._logger.info(f"Migration successful. Original DB backed up to {backup_db_path}")