feat(timeout): convert util/timeout to functional (#639)

Author: zhravan

cli/app/commands/install/command.py

@@ -2,7 +2,7 @@
 
 from app.utils.config import Config
 from app.utils.logger import create_logger, log_error, log_success, log_warning
-from app.utils.timeout import TimeoutWrapper
+from app.utils.timeout import timeout_wrapper
 
 from .deps import install_all_deps
 from .run import Install
@@ -211,7 +211,7 @@ def ssh(
         )
         ssh_operation = SSH(logger=logger)
 
-        with TimeoutWrapper(timeout):
+        with timeout_wrapper(timeout):
             result = ssh_operation.generate(config)
 
         log_success(result.output, verbose=verbose)
@@ -234,7 +234,7 @@ def deps(
     logger = create_logger(verbose=verbose)
     try:
 
-        with TimeoutWrapper(timeout):
+        with timeout_wrapper(timeout):
             result = install_all_deps(verbose=verbose, output=output, dry_run=dry_run)
 
         if output == "json":

cli/app/commands/install/development.py

@@ -33,7 +33,7 @@
 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
+from app.utils.timeout import timeout_wrapper
 
 from .base import BaseInstall
 from .deps import get_deps_from_config, get_installed_deps, install_dep
@@ -361,7 +361,7 @@ def _setup_clone_and_config(self):
             return
 
         try:
-            with TimeoutWrapper(self.timeout):
+            with timeout_wrapper(self.timeout):
                 success, error = clone_repository(
                     repo=self._get_config("repo_url"),
                     path=self._get_config("full_source_path"),
@@ -475,7 +475,7 @@ def _setup_ssh(self):
 
         ssh_operation = SSH(logger=self.logger)
         try:
-            with TimeoutWrapper(self.timeout):
+            with timeout_wrapper(self.timeout):
                 result = ssh_operation.generate(config)
         except TimeoutError:
             raise Exception(f"{ssh_setup_failed}: {operation_timed_out}")
@@ -548,7 +548,7 @@ def _load_proxy(self):
             return
 
         try:
-            with TimeoutWrapper(self.timeout):
+            with timeout_wrapper(self.timeout):
                 success, error = load_config(caddy_json_config, proxy_port, self.logger)
         except TimeoutError:
             raise Exception(f"Proxy load failed: {operation_timed_out}")
@@ -591,7 +591,7 @@ def _start_all_services(self):
 
         try:
             try:
-                with TimeoutWrapper(self.timeout):
+                with timeout_wrapper(self.timeout):
                     success, error = start_services(
                         name=self._get_config("service_name"),
                         detach=self._get_config("service_detach"),

cli/app/commands/install/run.py

@@ -42,7 +42,7 @@
 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
+from app.utils.timeout import timeout_wrapper
 
 from .deps import install_all_deps
 from .messages import (
@@ -306,7 +306,7 @@ def _run_preflight_checks(self):
 
     def _install_dependencies(self):
         try:
-            with TimeoutWrapper(self.timeout):
+            with timeout_wrapper(self.timeout):
                 result = install_all_deps(verbose=self.verbose, output="json", dry_run=self.dry_run)
         except TimeoutError:
             raise Exception(dependency_installation_timeout)
@@ -317,7 +317,7 @@ def _setup_clone_and_config(self):
             return
 
         try:
-            with TimeoutWrapper(self.timeout):
+            with timeout_wrapper(self.timeout):
                 success, error = clone_repository(
                     repo=self._get_config(DEFAULT_REPO),
                     path=self._get_config("full_source_path"),
@@ -417,7 +417,7 @@ def _setup_ssh(self):
         )
         ssh_operation = SSH(logger=self.logger)
         try:
-            with TimeoutWrapper(self.timeout):
+            with timeout_wrapper(self.timeout):
                 result = ssh_operation.generate(config)
         except TimeoutError:
             raise Exception(f"{ssh_setup_failed}: {operation_timed_out}")
@@ -455,7 +455,7 @@ def _start_services(self):
 
         try:
             try:
-                with TimeoutWrapper(self.timeout):
+                with timeout_wrapper(self.timeout):
                     success, error = start_services(
                         name="all",
                         detach=True,
@@ -490,7 +490,7 @@ def _load_proxy(self):
             return
 
         try:
-            with TimeoutWrapper(self.timeout):
+            with timeout_wrapper(self.timeout):
                 success, error = load_config(caddy_json_config, proxy_port, self.logger)
         except TimeoutError:
             raise Exception(f"{proxy_load_failed}: {operation_timed_out}")

cli/app/commands/uninstall/uninstall.py

@@ -8,7 +8,7 @@
 from app.utils.config import DEFAULT_COMPOSE_FILE, NIXOPUS_CONFIG_DIR, SSH_FILE_PATH, Config
 from app.utils.logger import create_logger
 from app.utils.protocols import LoggerProtocol
-from app.utils.timeout import TimeoutWrapper
+from app.utils.timeout import timeout_wrapper
 
 from .messages import (
     authorized_keys_not_found,
@@ -108,7 +108,7 @@ def stop_services_step(
         return True, None
 
     try:
-        with TimeoutWrapper(timeout):
+        with timeout_wrapper(timeout):
             success, error = stop_services(
                 name="all",
                 env_file=None,

cli/app/utils/timeout.py

@@ -1,26 +1,24 @@
 import signal
+from contextlib import contextmanager
+from typing import Generator
 
 from app.commands.install.messages import timeout_error
 
 
-class TimeoutWrapper:
-    """Context manager for timeout operations"""
+@contextmanager
+def timeout_wrapper(timeout: int) -> Generator[None, None, None]:
+    """Context manager for timeout operations using functional approach"""
+    if timeout > 0:
+        def timeout_handler(signum, frame):
+            raise TimeoutError(timeout_error.format(timeout=timeout))
 
-    def __init__(self, timeout: int):
-        self.timeout = timeout
-        self.original_handler = None
-
-    def __enter__(self):
-        if self.timeout > 0:
-
-            def timeout_handler(signum, frame):
-                raise TimeoutError(timeout_error.format(timeout=self.timeout))
-
-            self.original_handler = signal.signal(signal.SIGALRM, timeout_handler)
-            signal.alarm(self.timeout)
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        if self.timeout > 0:
+        original_handler = signal.signal(signal.SIGALRM, timeout_handler)
+        signal.alarm(timeout)
+        
+        try:
+            yield
+        finally:
             signal.alarm(0)
-            signal.signal(signal.SIGALRM, self.original_handler)
+            signal.signal(signal.SIGALRM, original_handler)
+    else:
+        yield