From 4a9b67a1cd03f8c9b6834ca30efab71276f25bd8 Mon Sep 17 00:00:00 2001 From: Collin Heist Date: Mon, 22 Jul 2024 17:46:39 -0600 Subject: [PATCH] [159] Create backup.py --- app/internal/backup.py | 185 ++++++++++++++++++++++++++++++++++++++ modules/ref/version_webui | 2 +- 2 files changed, 186 insertions(+), 1 deletion(-) create mode 100755 app/internal/backup.py diff --git a/app/internal/backup.py b/app/internal/backup.py new file mode 100755 index 000000000..868a78b9c --- /dev/null +++ b/app/internal/backup.py @@ -0,0 +1,185 @@ +from datetime import datetime, timedelta +from logging import Logger +from os import environ +from pathlib import Path +from shutil import copy as file_copy +from sqlite3 import connect +from typing import NamedTuple + +from app.schemas.preferences import DatabaseBackup, SettingsBackup, SystemBackup +from modules.Debug import log + + +BACKUP_DT_FORMAT = '%Y-%m-%d_%H-%M-%S' +BACKUP_RETENTION = timedelta(days=int(environ.get('TCM_BACKUP_RETENTION', 21))) +IS_DOCKER = environ.get('TCM_IS_DOCKER', 'false').lower() == 'true' + + +class DataBackup(NamedTuple): # pylint: disable=missing-class-docstring + config: Path + database: Path + + +def delete_old_backups( + backup_directory: Path, + *, + log: Logger = log, + ) -> None: + """ + Delete all old backups. + + Args: + backup_directory: Directory containing backup files to delete. + log: Logger for all log messages. + """ + + delete_before = datetime.now() - BACKUP_RETENTION + + for backup in backup_directory.iterdir(): + # Backup subdirectories + if backup.is_dir(): + try: + date = datetime.strptime(backup.name, BACKUP_DT_FORMAT) + except ValueError: + log.warning(f'Cannot identify date of backup file "{backup}"') + continue + + if date < delete_before: + for file in backup.iterdir(): + file.unlink(missing_ok=True) + log.debug(f'Deleted old backup "{backup.name}/{file.name}"') + backup.rmdir() + # Old-style files not stored in a subdirectory + else: + try: + date = datetime.strptime( + backup.name.rsplit('.')[-1], + BACKUP_DT_FORMAT + ) + except ValueError: + log.debug(f'Cannot identify date of backup file "{backup}"') + continue + + if date < delete_before: + backup.unlink(missing_ok=True) + log.debug(f'Deleted old backup "{backup}"') + + +def backup_data(version: str, *, log: Logger = log) -> DataBackup: + """ + Perform a backup of the SQL database and global preferences. + + Args: + version: Current version of TCM. + log: Logger for all log messages. + + Returns: + Tuple of Paths to created preferences and database backup files. + """ + + # Store backups in a dated subfolder + date = datetime.now().strftime(BACKUP_DT_FORMAT) + pre = '' if IS_DOCKER else '.' + backup_folder = Path(f'{pre}/config/backups/{date}') + backup_folder.mkdir(exist_ok=True, parents=True) + + # Identify source and destination files + config = Path(f'{pre}/config/config.pickle') + config_backup = backup_folder / f'config.pickle.{version}' + database = Path(f'{pre}/config/db.sqlite') + database_backup = backup_folder / f'db.sqlite.{version}' + + delete_old_backups(backup_folder.parent, log=log) + + # Backup config + if config.exists(): + file_copy(config, config_backup) + log.info(f'Performed settings backup ({config_backup})') + + # Backup database + if database.exists(): + file_copy(database, database_backup) + log.info(f'Performed database backup ({database_backup})') + + return DataBackup(config=config_backup, database=database_backup) + + +def restore_backup(backup: DataBackup, /, *, log: Logger = log): + """ + Restore the config and database from the given data backup. + + Args: + backup: Tuple of backup data (as returned by `backup_data()`) + to restore from. + log: Logger for all log messages. + """ + + # Restore config + if backup.config.exists(): + if IS_DOCKER: + file_copy(backup.config, Path('/config/config.pickle')) + else: + file_copy(backup.config, Path('./config/config.pickle')) + log.debug(f'Restored backup from "{backup.config}"') + else: + log.warning(f'Cannot restore backup from "{backup.config}"') + + # Restore database + if backup.database.exists(): + if IS_DOCKER: + file_copy(backup.database, Path('/config/db.sqlite')) + else: + file_copy(backup.database, Path('./config/db.sqlite')) + log.debug(f'Restored backup from "{backup.database}"') + else: + log.warning(f'Cannot restore backup from "{backup.database}"') + + +def list_available_backups() -> list[SystemBackup]: + """ + Get a list detailing all the available system backups. + + Returns: + List of system backup information. + """ + + backup_dir = Path('/config/backups' if IS_DOCKER else './config/backups') + + def _parse_version_number(file: Path) -> str: + """Parse the version number from the given file.""" + return file.name[len('config.pickle') + 1:] + + def _parse_schema_version(file: Path) -> str: + """Parse the alembic schema version from the given file.""" + connection = connect(file) + try: + return connection.cursor()\ + .execute('SELECT * FROM alembic_version LIMIT 1')\ + .fetchone()[0] + finally: + connection.close() + + backup_files = [ + ( + next(subfolder.glob('config.pickle*')), + next(subfolder.glob('db.sqlite*')) + ) + for subfolder in backup_dir.glob('2*') + ] + + return [ + SystemBackup( + database=DatabaseBackup( + filename=database.name, + filesize=database.stat().st_size, + schema_version=_parse_schema_version(database), + ), + settings=SettingsBackup( + filename=settings.name, + filesize=settings.stat().st_size, + ), + timestamp=datetime.strptime(settings.parent.name, BACKUP_DT_FORMAT), + version=_parse_version_number(settings), + ) + for settings, database in backup_files + ] diff --git a/modules/ref/version_webui b/modules/ref/version_webui index dc430ca42..00bc2b098 100755 --- a/modules/ref/version_webui +++ b/modules/ref/version_webui @@ -1 +1 @@ -v2.0-alpha.10.1-webui158 \ No newline at end of file +v2.0-alpha.10.1-webui159 \ No newline at end of file