feat(api-tools): add use_log and dry_run params to __execute_command function

Author: JeanExtreme002

src/api/services/apps.py

@@ -655,7 +655,7 @@ async def download_file(
 
     @staticmethod
     async def sync_dokku_with_api_database() -> None:
-        success, message = await run_command("apps:list")
+        success, message = await run_command("apps:list", use_log=False)
 
         if not success:
             logging.warning("Could not recover apps list to sync with database")
@@ -675,6 +675,6 @@ async def sync_dokku_with_api_database() -> None:
             logging.warning(
                 f"[sync_dokku_w_app_database]:{app_name}::Destroying unused application..."
             )
-            await run_command(f"--force apps:destroy {app_name}")
+            await run_command(f"--force apps:destroy {app_name}", use_log=False)
 
         logging.warning("[sync_dokku_w_app_database]::Sync complete.")

src/api/services/databases.py

@@ -389,7 +389,7 @@ async def sync_dokku_with_api_database() -> None:
         services = {}
 
         for plugin_name in available_databases:
-            success, message = await run_command(f"{plugin_name}:list")
+            success, message = await run_command(f"{plugin_name}:list", use_log=False)
 
             if not success:
                 logging.warning(
@@ -413,6 +413,8 @@ async def sync_dokku_with_api_database() -> None:
             logging.warning(
                 f"[sync_dokku_w_service_database]:{plugin_name}:{service_name}::Destroying unused service..."
             )
-            await run_command(f"--force {plugin_name}:destroy {service_name}")
+            await run_command(
+                f"--force {plugin_name}:destroy {service_name}", use_log=False
+            )
 
         logging.warning("[sync_dokku_w_service_database]::Sync complete.")

src/api/services/networks.py

@@ -182,7 +182,7 @@ async def get_linked_apps(
 
     @staticmethod
     async def sync_dokku_with_api_database() -> None:
-        success, message = await run_command("network:list")
+        success, message = await run_command("network:list", use_log=False)
 
         if not success:
             logging.warning("Could not recover networks list to sync with database")
@@ -206,6 +206,6 @@ async def sync_dokku_with_api_database() -> None:
             logging.warning(
                 f"[sync_dokku_w_network_database]:{network_name}::Destroying unused network..."
             )
-            await run_command(f"--force network:destroy {network_name}")
+            await run_command(f"--force network:destroy {network_name}", use_log=False)
 
         logging.warning("[sync_dokku_w_network_database]::Sync complete.")

src/api/tools/ssh.py

@@ -34,21 +34,29 @@ def get_command_history() -> list:
         return list(command_history)
 
 
-async def __execute_command(command: str, username: str) -> Tuple[bool, str]:
+async def __execute_command(
+    command: str, username: str, use_log: bool = True, dry_run: bool = False
+) -> Tuple[bool, str]:
     """
     Execute a command on the remote server via SSH.
 
     Args:
         command (str): The command to execute.
         username (str): The SSH username.
+        use_log (bool): If True, save the command at a command history (in-memory).
+        dry_run (bool): If True, the command is not actually executed.
     Returns:
         Tuple[bool, str]: A tuple containing a boolean indicating success
         or failure and the output or error message.
     """
     if username == "root":
         command = f"dokku {command}"
 
-    _log_command(command)
+    if use_log:
+        _log_command(command)
+
+    if dry_run:
+        return True, ""
 
     try:
         async with asyncssh.connect(
@@ -80,17 +88,23 @@ async def __execute_command(command: str, username: str) -> Tuple[bool, str]:
         return False, str(error)
 
 
-async def run_command(command: str) -> Tuple[bool, str]:
+async def run_command(
+    command: str, use_log: bool = True, dry_run: bool = False
+) -> Tuple[bool, str]:
     """
     Run a command on the remote server via SSH.
 
     Args:
         command (str): The command to execute.
+        use_log (bool): If True, save the command at a command history (in-memory).
+        dry_run (bool): If True, the command is not actually executed.
     Returns:
         Tuple[bool, str]: A tuple containing a boolean indicating success
         or failure and the output or error message.
     """
-    success, message = await __execute_command(command, "dokku")
+    success, message = await __execute_command(
+        command, "dokku", use_log=use_log, dry_run=dry_run
+    )
 
     if success:
         logging.info(f"Command executed successfully: {command}")
@@ -99,17 +113,23 @@ async def run_command(command: str) -> Tuple[bool, str]:
     return success, message
 
 
-async def run_command_as_root(command: str) -> Tuple[bool, str]:
+async def run_command_as_root(
+    command: str, use_log: bool = True, dry_run: bool = False
+) -> Tuple[bool, str]:
     """
     Run a command on the remote server via SSH as root.
 
     Args:
         command (str): The command to execute.
+        use_log (bool): If True, save the command at a command history (in-memory).
+        dry_run (bool): If True, the command is not actually executed.
     Returns:
         Tuple[bool, str]: A tuple containing a boolean indicating success
         or failure and the output or error message.
     """
-    success, message = await __execute_command(command, "root")
+    success, message = await __execute_command(
+        command, "root", use_log=use_log, dry_run=dry_run
+    )
 
     if success:
         logging.info(f"Command executed successfully: {command}")