refactor: improve component removal routines

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

kiauh/components/klipper/klipper_remove.py

@@ -7,14 +7,13 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 
-import shutil
 from typing import List, Union
 
 from components.klipper import KLIPPER_DIR, KLIPPER_ENV_DIR
 from components.klipper.klipper import Klipper
 from components.klipper.klipper_dialogs import print_instance_overview
 from core.instance_manager.instance_manager import InstanceManager
-from utils.fs_utils import remove_file
+from utils.fs_utils import run_remove_routines
 from utils.input_utils import get_selection_input
 from utils.logger import Logger
 from utils.sys_utils import cmd_sysctl_manage
@@ -49,10 +48,10 @@ def run_klipper_removal(
     else:
         if remove_dir:
             Logger.print_status("Removing Klipper local repository ...")
-            remove_klipper_dir()
+            run_remove_routines(KLIPPER_DIR)
         if remove_env:
             Logger.print_status("Removing Klipper Python environment ...")
-            remove_klipper_env()
+            run_remove_routines(KLIPPER_ENV_DIR)
 
     # delete klipper logs of all instances
     if delete_logs:
@@ -96,28 +95,6 @@ def remove_instances(
     cmd_sysctl_manage("daemon-reload")
 
 
-def remove_klipper_dir() -> None:
-    if not KLIPPER_DIR.exists():
-        Logger.print_info(f"'{KLIPPER_DIR}' does not exist. Skipped ...")
-        return
-
-    try:
-        shutil.rmtree(KLIPPER_DIR)
-    except OSError as e:
-        Logger.print_error(f"Unable to delete '{KLIPPER_DIR}':\n{e}")
-
-
-def remove_klipper_env() -> None:
-    if not KLIPPER_ENV_DIR.exists():
-        Logger.print_info(f"'{KLIPPER_ENV_DIR}' does not exist. Skipped ...")
-        return
-
-    try:
-        shutil.rmtree(KLIPPER_ENV_DIR)
-    except OSError as e:
-        Logger.print_error(f"Unable to delete '{KLIPPER_ENV_DIR}':\n{e}")
-
-
 def delete_klipper_logs(instances: List[Klipper]) -> None:
     all_logfiles = []
     for instance in instances:
@@ -128,4 +105,4 @@ def delete_klipper_logs(instances: List[Klipper]) -> None:
 
     for log in all_logfiles:
         Logger.print_status(f"Remove '{log}'")
-        remove_file(log)
+        run_remove_routines(log)

kiauh/components/moonraker/moonraker_remove.py

@@ -7,15 +7,14 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 
-import shutil
+from subprocess import DEVNULL, PIPE, CalledProcessError, run
-import subprocess
 from typing import List, Union
 
 from components.klipper.klipper_dialogs import print_instance_overview
 from components.moonraker import MOONRAKER_DIR, MOONRAKER_ENV_DIR
 from components.moonraker.moonraker import Moonraker
 from core.instance_manager.instance_manager import InstanceManager
-from utils.fs_utils import remove_file
+from utils.fs_utils import run_remove_routines
 from utils.input_utils import get_selection_input
 from utils.logger import Logger
 from utils.sys_utils import cmd_sysctl_manage
@@ -55,10 +54,10 @@ def run_moonraker_removal(
             remove_polkit_rules()
         if remove_dir:
             Logger.print_status("Removing Moonraker local repository ...")
-            remove_moonraker_dir()
+            run_remove_routines(MOONRAKER_DIR)
         if remove_env:
             Logger.print_status("Removing Moonraker Python environment ...")
-            remove_moonraker_env()
+            run_remove_routines(MOONRAKER_ENV_DIR)
 
     # delete moonraker logs of all instances
     if delete_logs:
@@ -102,28 +101,6 @@ def remove_instances(
     cmd_sysctl_manage("daemon-reload")
 
 
-def remove_moonraker_dir() -> None:
-    if not MOONRAKER_DIR.exists():
-        Logger.print_info(f"'{MOONRAKER_DIR}' does not exist. Skipped ...")
-        return
-
-    try:
-        shutil.rmtree(MOONRAKER_DIR)
-    except OSError as e:
-        Logger.print_error(f"Unable to delete '{MOONRAKER_DIR}':\n{e}")
-
-
-def remove_moonraker_env() -> None:
-    if not MOONRAKER_ENV_DIR.exists():
-        Logger.print_info(f"'{MOONRAKER_ENV_DIR}' does not exist. Skipped ...")
-        return
-
-    try:
-        shutil.rmtree(MOONRAKER_ENV_DIR)
-    except OSError as e:
-        Logger.print_error(f"Unable to delete '{MOONRAKER_ENV_DIR}':\n{e}")
-
-
 def remove_polkit_rules() -> None:
     if not MOONRAKER_DIR.exists():
         log = "Cannot remove policykit rules. Moonraker directory not found."
@@ -131,17 +108,9 @@ def remove_polkit_rules() -> None:
         return
 
     try:
-        command = [
+        cmd = [f"{MOONRAKER_DIR}/scripts/set-policykit-rules.sh", "--clear"]
-            f"{MOONRAKER_DIR}/scripts/set-policykit-rules.sh",
+        run(cmd, stderr=PIPE, stdout=DEVNULL, check=True)
-            "--clear",
+    except CalledProcessError as e:
-        ]
-        subprocess.run(
-            command,
-            stderr=subprocess.PIPE,
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-    except subprocess.CalledProcessError as e:
         Logger.print_error(f"Error while removing policykit rules: {e}")
 
     Logger.print_ok("Policykit rules successfully removed!")
@@ -157,4 +126,4 @@ def delete_moonraker_logs(instances: List[Moonraker]) -> None:
 
     for log in all_logfiles:
         Logger.print_status(f"Remove '{log}'")
-        remove_file(log)
+        run_remove_routines(log)

kiauh/components/webui_client/client_config/client_config_remove.py

@@ -8,8 +8,6 @@
 # ======================================================================= #
 
 
-import shutil
-import subprocess
 from typing import List
 
 from components.klipper.klipper import Klipper
@@ -17,7 +15,7 @@ from components.moonraker.moonraker import Moonraker
 from components.webui_client.base_data import BaseWebClientConfig
 from core.instance_manager.instance_manager import InstanceManager
 from utils.config_utils import remove_config_section
-from utils.fs_utils import remove_file
+from utils.fs_utils import run_remove_routines
 from utils.logger import Logger
 
 
@@ -33,29 +31,12 @@ def run_client_config_removal(
 
 
 def remove_client_config_dir(client_config: BaseWebClientConfig) -> None:
-    Logger.print_status(f"Removing {client_config.name} ...")
+    Logger.print_status(f"Removing {client_config.display_name} ...")
-    client_config_dir = client_config.config_dir
+    run_remove_routines(client_config.config_dir)
-    if not client_config_dir.exists():
-        Logger.print_info(f"'{client_config_dir}' does not exist. Skipping ...")
-        return
-
-    try:
-        shutil.rmtree(client_config_dir)
-    except OSError as e:
-        Logger.print_error(f"Unable to delete '{client_config_dir}':\n{e}")
 
 
 def remove_client_config_symlink(client_config: BaseWebClientConfig) -> None:
     im = InstanceManager(Klipper)
     instances: List[Klipper] = im.instances
     for instance in instances:
-        Logger.print_status(f"Removing symlink from '{instance.cfg_dir}' ...")
+        run_remove_routines(instance.cfg_dir.joinpath(client_config.config_filename))
-        symlink = instance.cfg_dir.joinpath(client_config.config_filename)
-        if not symlink.is_symlink():
-            Logger.print_info(f"'{symlink}' does not exist. Skipping ...")
-            continue
-
-        try:
-            remove_file(symlink)
-        except subprocess.CalledProcessError:
-            Logger.print_error("Failed to remove symlink!")

kiauh/components/webui_client/client_remove.py

@@ -8,7 +8,6 @@
 # ======================================================================= #
 
 
-import shutil
 from typing import List
 
 from components.klipper.klipper import Klipper
@@ -26,6 +25,7 @@ from utils.config_utils import remove_config_section
 from utils.fs_utils import (
     remove_nginx_config,
     remove_nginx_logs,
+    run_remove_routines,
 )
 from utils.logger import Logger
 
@@ -63,12 +63,4 @@ def run_client_removal(
 
 def remove_client_dir(client: BaseWebClient) -> None:
     Logger.print_status(f"Removing {client.display_name} ...")
-    client_dir = client.client_dir
+    run_remove_routines(client.client_dir)
-    if not client.client_dir.exists():
-        Logger.print_info(f"'{client_dir}' does not exist. Skipping ...")
-        return
-
-    try:
-        shutil.rmtree(client_dir)
-    except OSError as e:
-        Logger.print_error(f"Unable to delete '{client_dir}':\n{e}")

kiauh/components/webui_client/client_utils.py

@@ -74,7 +74,6 @@ def get_current_client_config(clients: List[BaseWebClient]) -> str:
 
 def backup_mainsail_config_json(is_temp=False) -> None:
     c_json = MainsailData().client_dir.joinpath("config.json")
-    Logger.print_status(f"Backup '{c_json}' ...")
     bm = BackupManager()
     if is_temp:
         fn = Path.home().joinpath("config.json.kiauh.bak")

kiauh/utils/fs_utils.py

@@ -237,27 +237,20 @@ def get_next_free_port(ports_in_use: List[int]) -> int:
 
 def remove_nginx_config(name: str) -> None:
     Logger.print_status(f"Removing NGINX config for {name.capitalize()} ...")
-    try:
-        remove_file(NGINX_SITES_AVAILABLE.joinpath(name), True)
-        remove_file(NGINX_SITES_ENABLED.joinpath(name), True)
 
-    except CalledProcessError as e:
+    run_remove_routines(NGINX_SITES_AVAILABLE.joinpath(name))
-        log = f"Unable to remove NGINX config '{name}':\n{e.stderr.decode()}"
+    run_remove_routines(NGINX_SITES_ENABLED.joinpath(name))
-        Logger.print_error(log)
 
 
 def remove_nginx_logs(name: str, instances: List[Klipper]) -> None:
     Logger.print_status(f"Removing NGINX logs for {name.capitalize()} ...")
-    try:
+
-        remove_file(Path(f"/var/log/nginx/{name}-access.log"), True)
+    run_remove_routines(Path(f"/var/log/nginx/{name}-access.log"))
-        remove_file(Path(f"/var/log/nginx/{name}-error.log"), True)
+    run_remove_routines(Path(f"/var/log/nginx/{name}-error.log"))
 
     if not instances:
         return
 
     for instance in instances:
-            remove_file(instance.log_dir.joinpath(f"{name}-access.log"))
+        run_remove_routines(instance.log_dir.joinpath(f"{name}-access.log"))
-            remove_file(instance.log_dir.joinpath(f"{name}-error.log"))
+        run_remove_routines(instance.log_dir.joinpath(f"{name}-error.log"))
-
-    except (OSError, CalledProcessError) as e:
-        Logger.print_error(f"Unable to remove NGINX logs:\n{e}")