refactor(BackupManager): rework backup structure and implement single file backup method
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
@@ -11,6 +11,7 @@
|
|||||||
|
|
||||||
import shutil
|
import shutil
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import List
|
||||||
|
|
||||||
from kiauh import KIAUH_BACKUP_DIR
|
from kiauh import KIAUH_BACKUP_DIR
|
||||||
from kiauh.utils.common import get_current_date
|
from kiauh.utils.common import get_current_date
|
||||||
@@ -19,51 +20,47 @@ from kiauh.utils.logger import Logger
|
|||||||
|
|
||||||
# noinspection PyMethodMayBeStatic
|
# noinspection PyMethodMayBeStatic
|
||||||
class BackupManager:
|
class BackupManager:
|
||||||
def __init__(
|
def __init__(self, backup_root_dir: Path = KIAUH_BACKUP_DIR):
|
||||||
self, backup_name: str, source: Path, backup_dir: Path = KIAUH_BACKUP_DIR
|
self._backup_root_dir = backup_root_dir
|
||||||
):
|
|
||||||
self._backup_name = backup_name
|
|
||||||
self._source = source
|
|
||||||
self._backup_dir = backup_dir
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def backup_name(self) -> str:
|
def backup_root_dir(self) -> Path:
|
||||||
return self._backup_name
|
return self._backup_root_dir
|
||||||
|
|
||||||
@backup_name.setter
|
@backup_root_dir.setter
|
||||||
def backup_name(self, value: str):
|
def backup_root_dir(self, value: Path):
|
||||||
self._backup_name = value
|
self._backup_root_dir = value
|
||||||
|
|
||||||
@property
|
def backup_file(self, files: List[Path] = None, target: Path = None):
|
||||||
def source(self) -> Path:
|
if not files:
|
||||||
return self._source
|
raise ValueError("Parameter 'files' cannot be None or an empty List!")
|
||||||
|
|
||||||
@source.setter
|
target = self.backup_root_dir if target is None else target
|
||||||
def source(self, value: Path):
|
for file in files:
|
||||||
self._source = value
|
if Path(file).is_file():
|
||||||
|
date = get_current_date().get("date")
|
||||||
|
time = get_current_date().get("time")
|
||||||
|
filename = f"{file.stem}-{date}-{time}{file.suffix}"
|
||||||
|
try:
|
||||||
|
Path(target).mkdir(exist_ok=True)
|
||||||
|
shutil.copyfile(file, Path(target, filename))
|
||||||
|
except OSError as e:
|
||||||
|
Logger.print_error(f"Unable to backup '{file}':\n{e}")
|
||||||
|
continue
|
||||||
|
|
||||||
@property
|
def backup_directory(self, name: str, source: Path, target: Path = None) -> None:
|
||||||
def backup_dir(self) -> Path:
|
if source is None or not Path(source).exists():
|
||||||
return self._backup_dir
|
|
||||||
|
|
||||||
@backup_dir.setter
|
|
||||||
def backup_dir(self, value: Path):
|
|
||||||
self._backup_dir = value
|
|
||||||
|
|
||||||
def backup(self) -> None:
|
|
||||||
if self._source is None or not Path(self._source).exists():
|
|
||||||
raise OSError
|
raise OSError
|
||||||
|
|
||||||
|
target = self.backup_root_dir if target is None else target
|
||||||
try:
|
try:
|
||||||
log = f"Creating backup of {self.backup_name} in {self.backup_dir} ..."
|
log = f"Creating backup of {name} in {target} ..."
|
||||||
Logger.print_status(log)
|
Logger.print_status(log)
|
||||||
date = get_current_date()
|
date = get_current_date().get("date")
|
||||||
dest = Path(
|
time = get_current_date().get("time")
|
||||||
f"{self.backup_dir}/{self.backup_name}/{date.get('date')}-{date.get('time')}"
|
shutil.copytree(source, Path(target, f"{name}-{date}-{time}"))
|
||||||
)
|
|
||||||
shutil.copytree(src=self.source, dst=dest)
|
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
Logger.print_error(f"Unable to backup source directory. Not exist.\n{e}")
|
Logger.print_error(f"Unable to backup directory '{source}':\n{e}")
|
||||||
return
|
return
|
||||||
|
|
||||||
Logger.print_ok("Backup successfull!")
|
Logger.print_ok("Backup successfull!")
|
||||||
|
|||||||
@@ -264,11 +264,9 @@ def update_klipper() -> None:
|
|||||||
|
|
||||||
cm = ConfigManager(cfg_file=KIAUH_CFG)
|
cm = ConfigManager(cfg_file=KIAUH_CFG)
|
||||||
if cm.get_value("kiauh", "backup_before_update"):
|
if cm.get_value("kiauh", "backup_before_update"):
|
||||||
backup_manager = BackupManager(source=KLIPPER_DIR, backup_name="klipper")
|
bm = BackupManager()
|
||||||
backup_manager.backup()
|
bm.backup_directory("klipper", KLIPPER_DIR)
|
||||||
backup_manager.backup_name = "klippy-env"
|
bm.backup_directory("klippy-env", KLIPPER_ENV_DIR)
|
||||||
backup_manager.source = KLIPPER_ENV_DIR
|
|
||||||
backup_manager.backup()
|
|
||||||
|
|
||||||
instance_manager = InstanceManager(Klipper)
|
instance_manager = InstanceManager(Klipper)
|
||||||
instance_manager.stop_all_instance()
|
instance_manager.stop_all_instance()
|
||||||
|
|||||||
@@ -290,11 +290,9 @@ def update_moonraker() -> None:
|
|||||||
|
|
||||||
cm = ConfigManager(cfg_file=KIAUH_CFG)
|
cm = ConfigManager(cfg_file=KIAUH_CFG)
|
||||||
if cm.get_value("kiauh", "backup_before_update"):
|
if cm.get_value("kiauh", "backup_before_update"):
|
||||||
backup_manager = BackupManager(source=MOONRAKER_DIR, backup_name="moonraker")
|
bm = BackupManager()
|
||||||
backup_manager.backup()
|
bm.backup_directory("moonraker", MOONRAKER_DIR)
|
||||||
backup_manager.backup_name = "moonraker-env"
|
bm.backup_directory("moonraker-env", MOONRAKER_ENV_DIR)
|
||||||
backup_manager.source = MOONRAKER_ENV_DIR
|
|
||||||
backup_manager.backup()
|
|
||||||
|
|
||||||
instance_manager = InstanceManager(Moonraker)
|
instance_manager = InstanceManager(Moonraker)
|
||||||
instance_manager.stop_all_instance()
|
instance_manager.stop_all_instance()
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
import subprocess
|
|
||||||
|
|
||||||
# ======================================================================= #
|
# ======================================================================= #
|
||||||
# Copyright (C) 2020 - 2023 Dominik Willner <th33xitus@gmail.com> #
|
# Copyright (C) 2020 - 2023 Dominik Willner <th33xitus@gmail.com> #
|
||||||
@@ -10,6 +9,7 @@ import subprocess
|
|||||||
# This file may be distributed under the terms of the GNU GPLv3 license #
|
# This file may be distributed under the terms of the GNU GPLv3 license #
|
||||||
# ======================================================================= #
|
# ======================================================================= #
|
||||||
|
|
||||||
|
import subprocess
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, Literal, List, Type
|
from typing import Dict, Literal, List, Type
|
||||||
@@ -33,8 +33,8 @@ def get_current_date() -> Dict[Literal["date", "time"], str]:
|
|||||||
:return: Dict holding a date and time key:value pair
|
:return: Dict holding a date and time key:value pair
|
||||||
"""
|
"""
|
||||||
now: datetime = datetime.today()
|
now: datetime = datetime.today()
|
||||||
date: str = now.strftime("%Y-%m-%d")
|
date: str = now.strftime("%Y%m%d")
|
||||||
time: str = now.strftime("%H-%M-%S")
|
time: str = now.strftime("%H%M%S")
|
||||||
|
|
||||||
return {"date": date, "time": time}
|
return {"date": date, "time": time}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user