refactor(utils): decompose lib.py into focused utility modules (#634)

Author: zhravan

cli/app/commands/clone/clone.py

@@ -2,7 +2,7 @@
 import subprocess
 from typing import Optional
 
-from app.utils.lib import DirectoryManager
+from app.utils.directory_manager import path_exists_and_not_force, remove_directory
 from app.utils.protocols import LoggerProtocol
 
 from .messages import (
@@ -60,10 +60,12 @@ def _build_clone_command(repo: str, path: str, branch: Optional[str] = None) ->
     return cmd
 
 
-def _execute_git_clone(repo: str, path: str, branch: Optional[str], logger: Optional[LoggerProtocol] = None) -> tuple[bool, Optional[str]]:
+def _execute_git_clone(
+    repo: str, path: str, branch: Optional[str], logger: Optional[LoggerProtocol] = None
+) -> tuple[bool, Optional[str]]:
     """Execute git clone command."""
     cmd = _build_clone_command(repo, path, branch)
-    
+
     if logger:
         logger.debug(debug_executing_git_clone.format(command=" ".join(cmd)))
 
@@ -89,8 +91,7 @@ def _prepare_target_directory(path: str, force: bool, logger: Optional[LoggerPro
     if force and os.path.exists(path):
         if logger:
             logger.debug(debug_removing_directory.format(path=path))
-        dir_manager = DirectoryManager()
-        success = dir_manager.remove_directory(path, logger)
+        success = remove_directory(path, logger)
         if not success and logger:
             logger.debug(debug_directory_removal_failed)
         return success
@@ -99,8 +100,7 @@ def _prepare_target_directory(path: str, force: bool, logger: Optional[LoggerPro
 
 def _validate_prerequisites(path: str, force: bool, logger: Optional[LoggerProtocol] = None) -> bool:
     """Validate prerequisites for cloning."""
-    dir_manager = DirectoryManager()
-    if dir_manager.path_exists_and_not_force(path, force):
+    if path_exists_and_not_force(path, force):
         if logger:
             logger.debug(debug_path_exists_force_disabled.format(path=path))
             logger.error(path_already_exists_use_force.format(path=path))
@@ -117,25 +117,25 @@ def clone_repository(
 ) -> tuple[bool, Optional[str]]:
     """Clone a git repository."""
     import time
-    
+
     start_time = time.time()
-    
+
     repo = _validate_repo(repo)
     path = _validate_path(path)
-    
+
     if logger:
         logger.debug(debug_cloning_repo.format(repo=repo, path=path, force=force))
-    
+
     if not _validate_prerequisites(path, force, logger):
         return False, prerequisites_validation_failed
-    
+
     if not _prepare_target_directory(path, force, logger):
         return False, failed_to_prepare_target_directory
-    
+
     success, error = _execute_git_clone(repo, path, branch, logger)
-    
+
     duration = time.time() - start_time
     if logger:
         logger.debug(debug_clone_completed.format(duration=f"{duration:.2f}", success=success))
-    
+
     return success, error

cli/app/commands/install/deps.py

@@ -3,7 +3,8 @@
 import subprocess
 
 from app.utils.config import DEPS, Config
-from app.utils.lib import HostInformation, ParallelProcessor
+from app.utils.host_information import get_os_name, get_package_manager
+from app.utils.parallel_processor import process_parallel
 from app.utils.logger import create_logger
 
 from .messages import (
@@ -122,8 +123,8 @@ def check_dependency(self, dep, package_manager):
 def install_all_deps(verbose=False, output="text", dry_run=False):
     logger = create_logger(verbose=verbose)
     deps = get_deps_from_config()
-    os_name = HostInformation.get_os_name()
-    package_manager = HostInformation.get_package_manager()
+    os_name = get_os_name()
+    package_manager = get_package_manager()
     if not package_manager:
         raise Exception(no_supported_package_manager)
     installed = get_installed_deps(deps, os_name, package_manager, verbose=verbose)
@@ -138,7 +139,7 @@ def error_handler(dep, exc):
         logger.error(f"Failed to install {dep['name']}: {exc}")
         return {"dependency": dep["name"], "installed": False}
 
-    results = ParallelProcessor.process_items(
+    results = process_parallel(
         to_install,
         install_wrapper,
         max_workers=min(len(to_install), 8),

cli/app/commands/install/development.py

@@ -29,7 +29,9 @@
     VIEW_ENV_FILE,
     VIEW_PORT,
 )
-from app.utils.lib import FileManager, HostInformation
+from app.utils.directory_manager import create_directory
+from app.utils.file_manager import get_directory_path, set_permissions
+from app.utils.host_information import get_os_name, get_package_manager
 from app.utils.protocols import LoggerProtocol
 from app.utils.timeout import TimeoutWrapper
 
@@ -323,8 +325,8 @@ def _run_preflight_checks(self):
     def _check_and_install_dependencies(self):
         """Check dependencies and install only if missing"""
         deps = get_deps_from_config()
-        os_name = HostInformation.get_os_name()
-        package_manager = HostInformation.get_package_manager()
+        os_name = get_os_name()
+        package_manager = get_package_manager()
 
         if not package_manager:
             raise Exception("No supported package manager found")
@@ -383,13 +385,13 @@ def _create_env_files(self):
             self.logger.info(f"  - View: {view_env_file}")
             return
 
-        FileManager.create_directory(FileManager.get_directory_path(api_env_file), logger=self.logger)
-        FileManager.create_directory(FileManager.get_directory_path(view_env_file), logger=self.logger)
+        create_directory(get_directory_path(api_env_file), logger=self.logger)
+        create_directory(get_directory_path(view_env_file), logger=self.logger)
 
         # Get combined env file path
         full_source_path = self._get_config("full_source_path")
         combined_env_file = os.path.join(full_source_path, ".env")
-        FileManager.create_directory(FileManager.get_directory_path(combined_env_file), logger=self.logger)
+        create_directory(get_directory_path(combined_env_file), logger=self.logger)
 
         services = [
             ("api", "services.api.env", api_env_file),
@@ -403,7 +405,7 @@ def _create_env_files(self):
             success, error = write_env_file(env_file, updated_env_values, self.logger)
             if not success:
                 raise Exception(f"{env_file_creation_failed} {service_name}: {error}")
-            file_perm_success, file_perm_error = FileManager.set_permissions(env_file, 0o644)
+            file_perm_success, file_perm_error = set_permissions(env_file, 0o644)
             if not file_perm_success:
                 raise Exception(f"{env_file_permissions_failed} {service_name}: {file_perm_error}")
             self.logger.debug(created_env_file.format(service_name=service_name, env_file=env_file))
@@ -420,7 +422,7 @@ def _create_env_files(self):
         if not success:
             raise Exception(f"{env_file_creation_failed} combined: {error}")
 
-        file_perm_success, file_perm_error = FileManager.set_permissions(combined_env_file, 0o644)
+        file_perm_success, file_perm_error = set_permissions(combined_env_file, 0o644)
         if not file_perm_success:
             raise Exception(f"{env_file_permissions_failed} combined: {file_perm_error}")
 
@@ -559,6 +561,12 @@ def _load_proxy(self):
 
     def _start_all_services(self):
         """Start all services (API, View, DB, Redis, Caddy) using Docker Compose"""
+        compose_file = self._get_config("compose_file_path")
+        
+        if self.dry_run:
+            self.logger.info(f"[DRY RUN] Would start services using {compose_file}")
+            return
+        
         env_vars = {}
         if self.api_port is not None:
             env_vars["API_PORT"] = str(self.api_port)
@@ -588,7 +596,7 @@ def _start_all_services(self):
                         name=self._get_config("service_name"),
                         detach=self._get_config("service_detach"),
                         env_file=None,
-                        compose_file=self._get_config("compose_file_path"),
+                        compose_file=compose_file,
                         logger=self.logger,
                     )
             except TimeoutError:

cli/app/commands/install/run.py

@@ -38,7 +38,9 @@
     VIEW_PORT,
     Config,
 )
-from app.utils.lib import FileManager, HostInformation
+from app.utils.directory_manager import create_directory
+from app.utils.file_manager import get_directory_path, set_permissions
+from app.utils.host_information import get_public_ip
 from app.utils.protocols import LoggerProtocol
 from app.utils.timeout import TimeoutWrapper
 
@@ -207,7 +209,7 @@ def _is_custom_repo_or_branch(self):
     def _get_host_ip(self) -> str:
         if self.host_ip:
             return self.host_ip
-        return HostInformation.get_public_ip()
+        return get_public_ip()
 
     def run(self):
         steps = [
@@ -334,9 +336,9 @@ def _create_env_files(self):
 
         full_source_path = self._get_config("full_source_path")
         combined_env_file = os.path.join(full_source_path, ".env")
-        FileManager.create_directory(FileManager.get_directory_path(api_env_file), logger=self.logger)
-        FileManager.create_directory(FileManager.get_directory_path(view_env_file), logger=self.logger)
-        FileManager.create_directory(FileManager.get_directory_path(combined_env_file), logger=self.logger)
+        create_directory(get_directory_path(api_env_file), logger=self.logger)
+        create_directory(get_directory_path(view_env_file), logger=self.logger)
+        create_directory(get_directory_path(combined_env_file), logger=self.logger)
 
         services = [
             ("api", "services.api.env", api_env_file),
@@ -348,7 +350,7 @@ def _create_env_files(self):
             success, error = write_env_file(env_file, updated_env_values, self.logger)
             if not success:
                 raise Exception(f"{env_file_creation_failed} {service_name}: {error}")
-            file_perm_success, file_perm_error = FileManager.set_permissions(env_file, 0o644)
+            file_perm_success, file_perm_error = set_permissions(env_file, 0o644)
             if not file_perm_success:
                 raise Exception(f"{env_file_permissions_failed} {service_name}: {file_perm_error}")
             self.logger.debug(created_env_file.format(service_name=service_name, env_file=env_file))
@@ -364,7 +366,7 @@ def _create_env_files(self):
 
         if not success:
             raise Exception(f"{env_file_creation_failed} combined: {error}")
-        file_perm_success, file_perm_error = FileManager.set_permissions(combined_env_file, 0o644)
+        file_perm_success, file_perm_error = set_permissions(combined_env_file, 0o644)
         if not file_perm_success:
             raise Exception(f"{env_file_permissions_failed} combined: {file_perm_error}")
         self.logger.debug(created_env_file.format(service_name="combined", env_file=combined_env_file))
@@ -423,6 +425,12 @@ def _setup_ssh(self):
             raise Exception(ssh_setup_failed)
 
     def _start_services(self):
+        compose_file = self._get_config("compose_file_path")
+        
+        if self.dry_run:
+            self.logger.info(f"[DRY RUN] Would start services using {compose_file}")
+            return
+        
         env_vars = {}
         if self.api_port is not None:
             env_vars["API_PORT"] = str(self.api_port)
@@ -452,7 +460,7 @@ def _start_services(self):
                         name="all",
                         detach=True,
                         env_file=None,
-                        compose_file=self._get_config("compose_file_path"),
+                        compose_file=compose_file,
                         logger=self.logger,
                     )
             except TimeoutError:
@@ -561,10 +569,10 @@ def _copy_caddyfile_to_target(self, full_source_path: str):
             source_caddyfile = os.path.join(full_source_path, "helpers", "Caddyfile")
             target_dir = _config.get(CADDY_CONFIG_VOLUME)
             target_caddyfile = os.path.join(target_dir, "Caddyfile")
-            FileManager.create_directory(target_dir, logger=self.logger)
+            create_directory(target_dir, logger=self.logger)
             if os.path.exists(source_caddyfile):
                 shutil.copy2(source_caddyfile, target_caddyfile)
-                FileManager.set_permissions(target_caddyfile, 0o644, logger=self.logger)
+                set_permissions(target_caddyfile, 0o644, logger=self.logger)
                 self.logger.debug(f"Copied Caddyfile from {source_caddyfile} to {target_caddyfile}")
             else:
                 self.logger.warning(f"Source Caddyfile not found at {source_caddyfile}")

cli/app/commands/install/ssh.py

@@ -5,7 +5,15 @@
 
 from pydantic import BaseModel, Field, field_validator
 
-from app.utils.lib import FileManager
+from app.utils.directory_manager import create_directory
+from app.utils.file_manager import (
+    append_to_file,
+    expand_user_path,
+    get_directory_path,
+    get_public_key_path,
+    read_file_content,
+    set_permissions,
+)
 from app.utils.logger import create_logger
 from app.utils.output_formatter import OutputFormatter
 from app.utils.protocols import LoggerProtocol
@@ -140,7 +148,6 @@ def format_dry_run(self, config: "SSHConfig") -> str:
 
 class SSHKeyManager:
     def __init__(self, logger: LoggerProtocol):
-        self.file_manager = FileManager()
         self.logger = logger
 
     def _check_ssh_keygen_availability(self) -> tuple[bool, str]:
@@ -201,15 +208,15 @@ def set_key_permissions(self, private_key_path: str, public_key_path: str) -> tu
         self.logger.debug(debug_ssh_permission_setting.format(private_key=private_key_path, public_key=public_key_path))
         try:
             self.logger.debug(debug_ssh_private_key_permissions.format(path=private_key_path))
-            private_success, private_error = self.file_manager.set_permissions(
+            private_success, private_error = set_permissions(
                 private_key_path, stat.S_IRUSR | stat.S_IWUSR, self.logger
             )
             if not private_success:
                 self.logger.debug(debug_ssh_private_key_permissions_failed.format(error=private_error))
                 return False, private_error
 
             self.logger.debug(debug_ssh_public_key_permissions.format(path=public_key_path))
-            public_success, public_error = self.file_manager.set_permissions(
+            public_success, public_error = set_permissions(
                 public_key_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH, self.logger
             )
             if not public_success:
@@ -227,7 +234,7 @@ def create_ssh_directory(self, ssh_dir: str) -> tuple[bool, str]:
         self.logger.debug(debug_ssh_directory_creation.format(directory=ssh_dir, permissions=oct(permissions)))
         try:
             self.logger.debug(debug_ssh_directory_check.format(directory=ssh_dir))
-            success, error = self.file_manager.create_directory(ssh_dir, permissions, self.logger)
+            success, error = create_directory(ssh_dir, permissions, self.logger)
             if success:
                 self.logger.debug(debug_ssh_directory_created.format(directory=ssh_dir))
             else:
@@ -242,12 +249,12 @@ def add_to_authorized_keys(self, public_key_path: str) -> tuple[bool, str]:
             self.logger.debug(adding_to_authorized_keys)
             self.logger.debug(debug_ssh_authorized_keys_read.format(path=public_key_path))
 
-            success, content, error = self.file_manager.read_file_content(public_key_path, self.logger)
+            success, content, error = read_file_content(public_key_path, self.logger)
             if not success:
                 self.logger.debug(debug_ssh_public_key_read_failed.format(error=error))
                 return False, error or failed_to_read_public_key
 
-            ssh_dir = self.file_manager.expand_user_path("~/.ssh")
+            ssh_dir = expand_user_path("~/.ssh")
             authorized_keys_path = os.path.join(ssh_dir, "authorized_keys")
             self.logger.debug(debug_ssh_authorized_keys_path.format(path=authorized_keys_path))
 
@@ -269,7 +276,7 @@ def add_to_authorized_keys(self, public_key_path: str) -> tuple[bool, str]:
                     return False, f"Failed to create authorized_keys file: {e}"
 
             self.logger.debug(debug_ssh_authorized_keys_append.format(path=authorized_keys_path))
-            success, error = self.file_manager.append_to_file(authorized_keys_path, content, self.logger)
+            success, error = append_to_file(authorized_keys_path, content, logger=self.logger)
             if not success:
                 self.logger.debug(debug_ssh_authorized_keys_append_failed.format(error=error))
                 return False, error or failed_to_append_to_authorized_keys
@@ -374,7 +381,6 @@ def __init__(self, config: SSHConfig, logger: LoggerProtocol = None, ssh_manager
         self.config = config
         self.ssh_manager = ssh_manager or SSHKeyManager(self.logger)
         self.formatter = SSHFormatter()
-        self.file_manager = FileManager()
 
     def _validate_prerequisites(self) -> bool:
         self.logger.debug(
@@ -383,7 +389,7 @@ def _validate_prerequisites(self) -> bool:
             )
         )
 
-        expanded_key_path = self.file_manager.expand_user_path(self.config.path)
+        expanded_key_path = expand_user_path(self.config.path)
         self.logger.debug(debug_ssh_path_expansion.format(original=self.config.path, expanded=expanded_key_path))
 
         if os.path.exists(expanded_key_path):
@@ -428,8 +434,8 @@ def generate_ssh_key(self) -> SSHResult:
             dry_run_output = self.formatter.format_dry_run(self.config)
             return self._create_result(True, dry_run_output)
 
-        expanded_path = self.file_manager.expand_user_path(self.config.path)
-        ssh_dir = self.file_manager.get_directory_path(expanded_path)
+        expanded_path = expand_user_path(self.config.path)
+        ssh_dir = get_directory_path(expanded_path)
         self.logger.debug(debug_ssh_key_directory_info.format(directory=ssh_dir))
 
         if self.config.create_ssh_directory:
@@ -449,7 +455,7 @@ def generate_ssh_key(self) -> SSHResult:
 
         if self.config.set_permissions:
             self.logger.debug(debug_ssh_permissions_enabled)
-            public_key_path = self.file_manager.get_public_key_path(expanded_path)
+            public_key_path = get_public_key_path(expanded_path)
             self.logger.debug(debug_ssh_public_key_path_info.format(path=public_key_path))
             success, error = self.ssh_manager.set_key_permissions(expanded_path, public_key_path)
             if not success:
@@ -458,7 +464,7 @@ def generate_ssh_key(self) -> SSHResult:
 
         if self.config.add_to_authorized_keys:
             self.logger.debug(debug_ssh_authorized_keys_enabled)
-            public_key_path = self.file_manager.get_public_key_path(expanded_path)
+            public_key_path = get_public_key_path(expanded_path)
             success, error = self.ssh_manager.add_to_authorized_keys(public_key_path)
             if not success:
                 self.logger.debug(debug_ssh_authorized_keys_failed_abort.format(error=error))