refactor: replace RepositoryManager by simple util functions

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

kiauh/components/klipper/klipper_setup.py

@@ -36,7 +36,7 @@ from components.klipper.klipper_utils import (
 )
 from components.moonraker.moonraker import Moonraker
 from core.instance_manager.instance_manager import InstanceManager
-from core.repo_manager.repo_manager import RepoManager
+from utils.git_utils import git_clone_wrapper, git_pull_wrapper
 from utils.input_utils import get_confirm
 from utils.logger import Logger
 from utils.system_utils import (
@@ -108,12 +108,10 @@ def install_klipper() -> None:
 
 def setup_klipper_prerequesites() -> None:
     settings = KiauhSettings()
-    repo_manager = RepoManager(
+    repo = settings.get("klipper", "repo_url")
-        repo=settings.get("klipper", "repo_url"),
+    branch = settings.get("klipper", "branch")
-        branch=settings.get("klipper", "branch"),
+
-        target_dir=KLIPPER_DIR,
+    git_clone_wrapper(repo, branch, KLIPPER_DIR)
-    )
-    repo_manager.clone_repo()
 
     # install klipper dependencies and create python virtualenv
     try:
@@ -152,12 +150,7 @@ def update_klipper() -> None:
     instance_manager = InstanceManager(Klipper)
     instance_manager.stop_all_instance()
 
-    repo_manager = RepoManager(
+    git_pull_wrapper(repo=settings.get("klipper", "repo_url"), target_dir=KLIPPER_DIR)
-        repo=settings.get("klipper", "repo_url"),
-        branch=settings.get("klipper", "branch"),
-        target_dir=KLIPPER_DIR,
-    )
-    repo_manager.pull_repo()
 
     # install possible new system packages
     install_klipper_packages(KLIPPER_DIR)

kiauh/components/klipper/klipper_utils.py

@@ -39,10 +39,10 @@ from core.config_manager.config_manager import ConfigManager
 from core.instance_manager.base_instance import BaseInstance
 from core.instance_manager.instance_manager import InstanceManager
 from core.instance_manager.name_scheme import NameScheme
-from core.repo_manager.repo_manager import RepoManager
 from utils import PRINTER_CFG_BACKUP_DIR
 from utils.common import get_install_status_common
 from utils.constants import CURRENT_USER
+from utils.git_utils import get_repo_name, get_remote_commit, get_local_commit
 from utils.input_utils import get_confirm, get_string_input, get_number_input
 from utils.logger import Logger
 from utils.system_utils import mask_system_service
@@ -59,9 +59,9 @@ def get_klipper_status() -> (
         "status": status.get("status"),
         "status_code": status.get("status_code"),
         "instances": status.get("instances"),
-        "repo": RepoManager.get_repo_name(KLIPPER_DIR),
+        "repo": get_repo_name(KLIPPER_DIR),
-        "local": RepoManager.get_local_commit(KLIPPER_DIR),
+        "local": get_local_commit(KLIPPER_DIR),
-        "remote": RepoManager.get_remote_commit(KLIPPER_DIR),
+        "remote": get_remote_commit(KLIPPER_DIR),
     }
 
 

kiauh/components/moonraker/moonraker_setup.py

@@ -35,8 +35,8 @@ from components.moonraker.moonraker_utils import (
     backup_moonraker_dir,
 )
 from core.instance_manager.instance_manager import InstanceManager
-from core.repo_manager.repo_manager import RepoManager
 from utils.filesystem_utils import check_file_exist
+from utils.git_utils import git_clone_wrapper, git_pull_wrapper
 from utils.input_utils import (
     get_confirm,
     get_selection_input,
@@ -135,12 +135,7 @@ def setup_moonraker_prerequesites() -> None:
     repo = settings.get("moonraker", "repo_url")
     branch = settings.get("moonraker", "branch")
 
-    repo_manager = RepoManager(
+    git_clone_wrapper(repo, branch, MOONRAKER_DIR)
-        repo=repo,
-        branch=branch,
-        target_dir=MOONRAKER_DIR,
-    )
-    repo_manager.clone_repo()
 
     # install moonraker dependencies and create python virtualenv
     install_moonraker_packages(MOONRAKER_DIR)
@@ -196,12 +191,9 @@ def update_moonraker() -> None:
     instance_manager = InstanceManager(Moonraker)
     instance_manager.stop_all_instance()
 
-    repo_manager = RepoManager(
+    git_pull_wrapper(
-        repo=settings.get("moonraker", "repo_url"),
+        repo=settings.get("moonraker", "repo_url"), target_dir=MOONRAKER_DIR
-        branch=settings.get("moonraker", "branch"),
-        target_dir=MOONRAKER_DIR,
     )
-    repo_manager.pull_repo()
 
     # install possible new system packages
     install_moonraker_packages(MOONRAKER_DIR)

kiauh/components/moonraker/moonraker_utils.py

@@ -25,8 +25,8 @@ from components.webui_client.mainsail_data import MainsailData
 from core.backup_manager.backup_manager import BackupManager
 from core.config_manager.config_manager import ConfigManager
 from core.instance_manager.instance_manager import InstanceManager
-from core.repo_manager.repo_manager import RepoManager
 from utils.common import get_install_status_common
+from utils.git_utils import get_repo_name, get_local_commit, get_remote_commit
 from utils.logger import Logger
 from utils.system_utils import (
     get_ipv4_addr,
@@ -44,9 +44,9 @@ def get_moonraker_status() -> (
         "status": status.get("status"),
         "status_code": status.get("status_code"),
         "instances": status.get("instances"),
-        "repo": RepoManager.get_repo_name(MOONRAKER_DIR),
+        "repo": get_repo_name(MOONRAKER_DIR),
-        "local": RepoManager.get_local_commit(MOONRAKER_DIR),
+        "local": get_local_commit(MOONRAKER_DIR),
-        "remote": RepoManager.get_remote_commit(MOONRAKER_DIR),
+        "remote": get_remote_commit(MOONRAKER_DIR),
     }
 
 

kiauh/components/webui_client/client_config/client_config_setup.py

@@ -25,13 +25,13 @@ from components.webui_client.client_utils import (
 )
 
 from core.instance_manager.instance_manager import InstanceManager
-from core.repo_manager.repo_manager import RepoManager
 from utils.common import backup_printer_config_dir
 from utils.filesystem_utils import (
     create_symlink,
     add_config_section,
     add_config_section_at_top,
 )
+from utils.git_utils import git_clone_wrapper, git_pull_wrapper
 from utils.input_utils import get_confirm
 from utils.logger import Logger
 
@@ -86,11 +86,9 @@ def install_client_config(client_data: BaseWebClient) -> None:
 def download_client_config(client_config: BaseWebClientConfig) -> None:
     try:
         Logger.print_status(f"Downloading {client_config.display_name} ...")
-        rm = RepoManager(
+        repo = client_config.repo_url
-            client_config.repo_url,
+        target_dir = client_config.config_dir
-            target_dir=str(client_config.config_dir),
+        git_clone_wrapper(repo, None, target_dir)
-        )
-        rm.clone_repo()
     except Exception:
         Logger.print_error(f"Downloading {client_config.display_name} failed!")
         raise
@@ -111,12 +109,7 @@ def update_client_config(client: BaseWebClient) -> None:
     if settings.get("kiauh", "backup_before_update"):
         backup_client_config_data(client)
 
-    repo_manager = RepoManager(
+    git_pull_wrapper(client_config.repo_url, client_config.config_dir)
-        repo=client_config.repo_url,
-        branch="master",
-        target_dir=str(client_config.config_dir),
-    )
-    repo_manager.pull_repo()
 
     Logger.print_ok(f"Successfully updated {client_config.display_name}.")
     Logger.print_info("Restart Klipper to reload the configuration!")

kiauh/components/webui_client/client_utils.py

@@ -21,12 +21,17 @@ from components.webui_client.base_data import (
 )
 from components.webui_client.mainsail_data import MainsailData
 from core.backup_manager.backup_manager import BackupManager
-from core.repo_manager.repo_manager import RepoManager
 from core.settings.kiauh_settings import KiauhSettings
 from utils import NGINX_SITES_AVAILABLE, NGINX_CONFD
 from utils.common import get_install_status_webui
 from utils.constants import COLOR_CYAN, RESET_FORMAT, COLOR_YELLOW
-from utils.git_utils import get_latest_tag, get_latest_unstable_tag
+from utils.git_utils import (
+    get_latest_tag,
+    get_latest_unstable_tag,
+    get_repo_name,
+    get_local_commit,
+    get_remote_commit,
+)
 from utils.logger import Logger
 
 
@@ -48,9 +53,9 @@ def get_client_config_status(
     client_config = client.client_config.config_dir
 
     return {
-        "repo": RepoManager.get_repo_name(client_config),
+        "repo": get_repo_name(client_config),
-        "local": RepoManager.get_local_commit(client_config),
+        "local": get_local_commit(client_config),
-        "remote": RepoManager.get_remote_commit(client_config),
+        "remote": get_remote_commit(client_config),
     }
 
 

kiauh/core/menus/settings_menu.py

@@ -18,9 +18,9 @@ from components.moonraker.moonraker import Moonraker
 from core.instance_manager.instance_manager import InstanceManager
 from core.menus import Option
 from core.menus.base_menu import BaseMenu
-from core.repo_manager.repo_manager import RepoManager
 from core.settings.kiauh_settings import KiauhSettings
 from utils.constants import COLOR_CYAN, RESET_FORMAT, COLOR_GREEN, COLOR_YELLOW
+from utils.git_utils import git_clone_wrapper
 from utils.input_utils import get_string_input, get_confirm
 from utils.logger import Logger
 
@@ -199,8 +199,7 @@ class SettingsMenu(BaseMenu):
 
         repo = self.kiauh_settings.get(name, "repo_url")
         branch = self.kiauh_settings.get(name, "branch")
-        repman = RepoManager(repo, str(target_dir), branch)
+        git_clone_wrapper(repo, branch, target_dir)
-        repman.clone_repo()
 
         im.start_all_instance()
 

kiauh/core/repo_manager/repo_manager.py

@@ -1,171 +0,0 @@
-# ======================================================================= #
-#  Copyright (C) 2020 - 2024 Dominik Willner <th33xitus@gmail.com>        #
-#                                                                         #
-#  This file is part of KIAUH - Klipper Installation And Update Helper    #
-#  https://github.com/dw-0/kiauh                                          #
-#                                                                         #
-#  This file may be distributed under the terms of the GNU GPLv3 license  #
-# ======================================================================= #
-
-import shutil
-import subprocess
-from pathlib import Path
-
-from utils.input_utils import get_confirm
-from utils.logger import Logger
-
-
-# noinspection PyMethodMayBeStatic
-class RepoManager:
-    def __init__(
-        self,
-        repo: str,
-        target_dir: str,
-        branch: str = None,
-    ):
-        self._repo = repo
-        self._branch = branch
-        self._method = self._get_method()
-        self._target_dir = target_dir
-
-    @property
-    def repo(self) -> str:
-        return self._repo
-
-    @repo.setter
-    def repo(self, value) -> None:
-        self._repo = value
-
-    @property
-    def branch(self) -> str:
-        return self._branch
-
-    @branch.setter
-    def branch(self, value) -> None:
-        self._branch = value
-
-    @property
-    def method(self) -> str:
-        return self._method
-
-    @method.setter
-    def method(self, value) -> None:
-        self._method = value
-
-    @property
-    def target_dir(self) -> str:
-        return self._target_dir
-
-    @target_dir.setter
-    def target_dir(self, value) -> None:
-        self._target_dir = value
-
-    @staticmethod
-    def get_repo_name(repo: Path) -> str:
-        """
-        Helper method to extract the organisation and name of a repository |
-        :param repo: repository to extract the values from
-        :return: String in form of "<orga>/<name>"
-        """
-        if not repo.exists() and not repo.joinpath(".git").exists():
-            return "-"
-
-        try:
-            cmd = ["git", "-C", repo, "config", "--get", "remote.origin.url"]
-            result = subprocess.check_output(cmd, stderr=subprocess.DEVNULL)
-            return "/".join(result.decode().strip().split("/")[-2:])
-        except subprocess.CalledProcessError:
-            return "-"
-
-    @staticmethod
-    def get_local_commit(repo: Path) -> str:
-        if not repo.exists() and not repo.joinpath(".git").exists():
-            return "-"
-
-        try:
-            cmd = f"cd {repo} && git describe HEAD --always --tags | cut -d '-' -f 1,2"
-            return subprocess.check_output(cmd, shell=True, text=True).strip()
-        except subprocess.CalledProcessError:
-            return "-"
-
-    @staticmethod
-    def get_remote_commit(repo: Path) -> str:
-        if not repo.exists() and not repo.joinpath(".git").exists():
-            return "-"
-
-        try:
-            # get locally checked out branch
-            branch_cmd = f"cd {repo} && git branch | grep -E '\*'"
-            branch = subprocess.check_output(branch_cmd, shell=True, text=True)
-            branch = branch.split("*")[-1].strip()
-            cmd = f"cd {repo} && git describe 'origin/{branch}' --always --tags | cut -d '-' -f 1,2"
-            return subprocess.check_output(cmd, shell=True, text=True).strip()
-        except subprocess.CalledProcessError:
-            return "-"
-
-    def clone_repo(self):
-        log = f"Cloning repository from '{self.repo}' with method '{self.method}'"
-        Logger.print_status(log)
-        try:
-            if Path(self.target_dir).exists():
-                question = f"'{self.target_dir}' already exists. Overwrite?"
-                if not get_confirm(question, default_choice=False):
-                    Logger.print_info("Skip cloning of repository ...")
-                    return
-                shutil.rmtree(self.target_dir)
-
-            self._clone()
-            self._checkout()
-        except subprocess.CalledProcessError:
-            log = "An unexpected error occured during cloning of the repository."
-            Logger.print_error(log)
-            return
-        except OSError as e:
-            Logger.print_error(f"Error removing existing repository: {e.strerror}")
-            return
-
-    def pull_repo(self) -> None:
-        Logger.print_status(f"Updating repository '{self.repo}' ...")
-        try:
-            self._pull()
-        except subprocess.CalledProcessError:
-            log = "An unexpected error occured during updating the repository."
-            Logger.print_error(log)
-            return
-
-    def _clone(self):
-        try:
-            command = ["git", "clone", self.repo, self.target_dir]
-            subprocess.run(command, check=True)
-
-            Logger.print_ok("Clone successful!")
-        except subprocess.CalledProcessError as e:
-            log = f"Error cloning repository {self.repo}: {e.stderr.decode()}"
-            Logger.print_error(log)
-            raise
-
-    def _checkout(self):
-        if self.branch is None:
-            return
-
-        try:
-            command = ["git", "checkout", f"{self.branch}"]
-            subprocess.run(command, cwd=self.target_dir, check=True)
-
-            Logger.print_ok("Checkout successful!")
-        except subprocess.CalledProcessError as e:
-            log = f"Error checking out branch {self.branch}: {e.stderr.decode()}"
-            Logger.print_error(log)
-            raise
-
-    def _pull(self) -> None:
-        try:
-            command = ["git", "pull"]
-            subprocess.run(command, cwd=self.target_dir, check=True)
-        except subprocess.CalledProcessError as e:
-            log = f"Error on git pull: {e.stderr.decode()}"
-            Logger.print_error(log)
-            raise
-
-    def _get_method(self) -> str:
-        return "ssh" if self.repo.startswith("git") else "https"

kiauh/extensions/mainsail_theme_installer/mainsail_theme_installer_extension.py

@@ -23,8 +23,8 @@ from extensions.base_extension import BaseExtension
 from core.instance_manager.base_instance import BaseInstance
 from core.instance_manager.instance_manager import InstanceManager
 from core.menus.base_menu import BaseMenu
-from core.repo_manager.repo_manager import RepoManager
 from utils.constants import COLOR_YELLOW, COLOR_CYAN, RESET_FORMAT
+from utils.git_utils import git_clone_wrapper
 from utils.input_utils import get_selection_input
 from utils.logger import Logger
 
@@ -136,10 +136,8 @@ class MainsailThemeInstallMenu(BaseMenu):
         if printer_list is None:
             return
 
-        repo_manager = RepoManager(theme_repo_url, "")
         for printer in printer_list:
-            repo_manager.target_dir = printer.cfg_dir.joinpath(".theme")
+            git_clone_wrapper(theme_repo_url, None, printer.cfg_dir.joinpath(".theme"))
-            repo_manager.clone_repo()
 
         if len(theme_data.get("short_note", "")) > 1:
             Logger.print_warn("Info from the creator:", prefix=False, start="\n")

kiauh/utils/git_utils.py

@@ -1,9 +1,11 @@
 import json
+import shutil
 import urllib.request
 from http.client import HTTPResponse
 from json import JSONDecodeError
-from subprocess import CalledProcessError, PIPE, run
+from pathlib import Path
-from typing import List, Type
+from subprocess import CalledProcessError, PIPE, run, check_output, DEVNULL
+from typing import List, Type, Optional
 
 from core.instance_manager.base_instance import BaseInstance
 from core.instance_manager.instance_manager import InstanceManager
@@ -11,6 +13,70 @@ from utils.input_utils import get_number_input, get_confirm
 from utils.logger import Logger
 
 
+def git_clone_wrapper(repo: str, branch: Optional[str], target_dir: Path) -> None:
+    """
+    Clones a repository from the given URL and checks out the specified branch if given.
+
+    :param repo: The URL of the repository to clone.
+    :param branch: The branch to check out. If None, the default branch will be checked out.
+    :param target_dir: The directory where the repository will be cloned.
+    :return: None
+    """
+    log = f"Cloning repository from '{repo}'"
+    Logger.print_status(log)
+    try:
+        if Path(target_dir).exists():
+            question = f"'{target_dir}' already exists. Overwrite?"
+            if not get_confirm(question, default_choice=False):
+                Logger.print_info("Skip cloning of repository ...")
+                return
+            shutil.rmtree(target_dir)
+
+        git_cmd_clone(repo, target_dir)
+        git_cmd_checkout(branch, target_dir)
+    except CalledProcessError:
+        log = "An unexpected error occured during cloning of the repository."
+        Logger.print_error(log)
+        return
+    except OSError as e:
+        Logger.print_error(f"Error removing existing repository: {e.strerror}")
+        return
+
+
+def git_pull_wrapper(repo: str, target_dir: Path) -> None:
+    """
+    A function that updates a repository using git pull.
+
+    :param repo: The repository to update.
+    :param target_dir: The directory of the repository.
+    :return: None
+    """
+    Logger.print_status(f"Updating repository '{repo}' ...")
+    try:
+        git_cmd_pull(target_dir)
+    except CalledProcessError:
+        log = "An unexpected error occured during updating the repository."
+        Logger.print_error(log)
+        return
+
+
+def get_repo_name(repo: Path) -> str:
+    """
+    Helper method to extract the organisation and name of a repository |
+    :param repo: repository to extract the values from
+    :return: String in form of "<orga>/<name>"
+    """
+    if not repo.exists() and not repo.joinpath(".git").exists():
+        return "-"
+
+    try:
+        cmd = ["git", "-C", repo, "config", "--get", "remote.origin.url"]
+        result = check_output(cmd, stderr=DEVNULL)
+        return "/".join(result.decode().strip().split("/")[-2:])
+    except CalledProcessError:
+        return "-"
+
+
 def get_tags(repo_path: str) -> List[str]:
     try:
         url = f"https://api.github.com/repos/{repo_path}/tags"
@@ -61,7 +127,70 @@ def get_latest_unstable_tag(repo_path: str) -> str:
         raise
 
 
-def rollback_repository(repo_dir: str, instance: Type[BaseInstance]) -> None:
+def get_local_commit(repo: Path) -> str:
+    if not repo.exists() and not repo.joinpath(".git").exists():
+        return "-"
+
+    try:
+        cmd = f"cd {repo} && git describe HEAD --always --tags | cut -d '-' -f 1,2"
+        return check_output(cmd, shell=True, text=True).strip()
+    except CalledProcessError:
+        return "-"
+
+
+def get_remote_commit(repo: Path) -> str:
+    if not repo.exists() and not repo.joinpath(".git").exists():
+        return "-"
+
+    try:
+        # get locally checked out branch
+        branch_cmd = f"cd {repo} && git branch | grep -E '\*'"
+        branch = check_output(branch_cmd, shell=True, text=True)
+        branch = branch.split("*")[-1].strip()
+        cmd = f"cd {repo} && git describe 'origin/{branch}' --always --tags | cut -d '-' -f 1,2"
+        return check_output(cmd, shell=True, text=True).strip()
+    except CalledProcessError:
+        return "-"
+
+
+def git_cmd_clone(repo: str, target_dir: Path) -> None:
+    try:
+        command = ["git", "clone", repo, target_dir]
+        run(command, check=True)
+
+        Logger.print_ok("Clone successful!")
+    except CalledProcessError as e:
+        log = f"Error cloning repository {repo}: {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise
+
+
+def git_cmd_checkout(branch: str, target_dir: Path) -> None:
+    if branch is None:
+        return
+
+    try:
+        command = ["git", "checkout", f"{branch}"]
+        run(command, cwd=target_dir, check=True)
+
+        Logger.print_ok("Checkout successful!")
+    except CalledProcessError as e:
+        log = f"Error checking out branch {branch}: {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise
+
+
+def git_cmd_pull(target_dir: Path) -> None:
+    try:
+        command = ["git", "pull"]
+        run(command, cwd=target_dir, check=True)
+    except CalledProcessError as e:
+        log = f"Error on git pull: {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise
+
+
+def rollback_repository(repo_dir: Path, instance: Type[BaseInstance]) -> None:
     q1 = "How many commits do you want to roll back"
     amount = get_number_input(q1, 1, allow_go_back=True)