Merge pull request #309 from tcdent/init-qol

Re-prompt user for project name instead of exiting, spinner for package install

Author: bboynton97

agentstack/cli/init.py

@@ -36,6 +36,41 @@ def require_uv():
         raise EnvironmentError(message)
 
 
+def prompt_slug_name() -> str:
+    """Prompt the user for a project name."""
+    
+    def _validate(slug_name: Optional[str]) -> bool:
+        if not slug_name:
+            log.error("Project name cannot be empty")
+            return False
+            
+        if not is_snake_case(slug_name):
+            log.error("Project name must be snake_case")
+            return False
+
+        if os.path.exists(conf.PATH / slug_name):
+            log.error(f"Project path already exists: {conf.PATH / slug_name}")
+            return False
+
+        return True
+    
+    def _prompt() -> str:
+        return inquirer.text(
+            message="Project name (snake_case)",
+        )
+    
+    log.info(
+        "Provide a project name. This will be used to create a new directory in the "
+        "current path and will be used as the project name. 🐍 Must be snake_case."
+    )
+    slug_name = None
+    while not _validate(slug_name):
+        slug_name = _prompt()
+    
+    assert slug_name  # appease type checker
+    return slug_name
+
+
 def select_template(slug_name: str, framework: Optional[str] = None) -> TemplateConfig:
     """Let the user select a template from the ones available."""
     templates: list[TemplateConfig] = get_all_templates()
@@ -85,22 +120,11 @@ def init_project(
     welcome_message()
 
     if not slug_name:
-        log.info(
-            "Provide a project name. This will be used to create a new directory in the "
-            "current path and will be used as the project name. 🐍 Must be snake_case."
-        )
-        slug_name = inquirer.text(
-            message="Project name (snake_case)",
-        )
-
-    if not slug_name:
-        raise Exception("Project name cannot be empty")
-    if not is_snake_case(slug_name):
-        raise Exception("Project name must be snake_case")
+        slug_name = prompt_slug_name()
 
     conf.set_path(conf.PATH / slug_name)
-    if os.path.exists(conf.PATH):  # cookiecutter requires the directory to not exist
-        raise Exception(f"Directory already exists: {conf.PATH}")
+    # cookiecutter requires the directory to not exist
+    assert not os.path.exists(conf.PATH), f"Directory already exists: {conf.PATH}"
 
     if use_wizard:
         log.debug("Initializing new project with wizard.")

agentstack/cli/spinner.py

@@ -0,0 +1,95 @@
+import itertools
+import shutil
+import sys
+import threading
+import time
+from typing import Optional, Literal
+
+from agentstack import log
+
+
+class Spinner:
+    def __init__(self, message="Working", delay=0.1):
+        self.spinner = itertools.cycle(['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'])
+        self.delay = delay
+        self.message = message
+        self.running = False
+        self.spinner_thread = None
+        self.start_time = None
+        self._lock = threading.Lock()
+        self._last_printed_len = 0
+        self._last_message = ""
+
+    def _clear_line(self):
+        """Clear the current line in terminal."""
+        sys.stdout.write('\r' + ' ' * self._last_printed_len + '\r')
+        sys.stdout.flush()
+
+    def spin(self):
+        while self.running:
+            with self._lock:
+                elapsed = time.time() - self.start_time
+                terminal_width = shutil.get_terminal_size().columns
+                spinner_char = next(self.spinner)
+                time_str = f"{elapsed:.1f}s"
+
+                # Format: [spinner] Message... [time]
+                message = f"\r{spinner_char} {self.message}... [{time_str}]"
+
+                # Ensure we don't exceed terminal width
+                if len(message) > terminal_width:
+                    message = message[:terminal_width - 3] + "..."
+
+                # Clear previous line and print new one
+                self._clear_line()
+                sys.stdout.write(message)
+                sys.stdout.flush()
+                self._last_printed_len = len(message)
+
+            time.sleep(self.delay)
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+
+    def start(self):
+        if not self.running:
+            self.running = True
+            self.start_time = time.time()
+            self.spinner_thread = threading.Thread(target=self.spin)
+            self.spinner_thread.start()
+
+    def stop(self):
+        if self.running:
+            self.running = False
+            if self.spinner_thread:
+                self.spinner_thread.join()
+            with self._lock:
+                self._clear_line()
+
+    def update_message(self, message):
+        """Update spinner message and ensure clean line."""
+        with self._lock:
+            self._clear_line()
+            self.message = message
+
+    def clear_and_log(self, message, color: Literal['success', 'info'] = 'success'):
+        """Temporarily clear spinner, print message, and resume spinner.
+        Skips printing if message is the same as the last message printed."""
+        with self._lock:
+            # Skip if message is same as last one
+            if hasattr(self, '_last_message') and self._last_message == message:
+                return
+
+            self._clear_line()
+            if color == 'success':
+                log.success(message)
+            else:
+                log.info(message)
+            sys.stdout.flush()
+
+            # Store current message
+            self._last_message = message

agentstack/packaging.py

@@ -24,44 +24,49 @@
 def install(package: str):
     """Install a package with `uv` and add it to pyproject.toml."""
 
+    from agentstack.cli.spinner import Spinner
+
     def on_progress(line: str):
         if RE_UV_PROGRESS.match(line):
-            log.info(line.strip())
+            spinner.clear_and_log(line.strip(), 'info')
 
     def on_error(line: str):
         log.error(f"uv: [error]\n {line.strip()}")
-
-    log.info(f"Installing {package}")
-    _wrap_command_with_callbacks(
-        [get_uv_bin(), 'add', '--python', '.venv/bin/python', package],
-        on_progress=on_progress,
-        on_error=on_error,
-    )
+    
+    with Spinner(f"Installing {package}") as spinner:
+        _wrap_command_with_callbacks(
+            [get_uv_bin(), 'add', '--python', '.venv/bin/python', package],
+            on_progress=on_progress,
+            on_error=on_error,
+        )
 
 
 def install_project():
     """Install all dependencies for the user's project."""
 
+    from agentstack.cli.spinner import Spinner
+
     def on_progress(line: str):
         if RE_UV_PROGRESS.match(line):
-            log.info(line.strip())
+            spinner.clear_and_log(line.strip(), 'info')
 
     def on_error(line: str):
         log.error(f"uv: [error]\n {line.strip()}")
 
     try:
-        result = _wrap_command_with_callbacks(
-            [get_uv_bin(), 'pip', 'install', '--python', '.venv/bin/python', '.'],
-            on_progress=on_progress,
-            on_error=on_error,
-        )
-        if result is False:
-            log.info("Retrying uv installation with --no-cache flag...")
-            _wrap_command_with_callbacks(
-                [get_uv_bin(), 'pip', 'install', '--no-cache', '--python', '.venv/bin/python', '.'],
+        with Spinner(f"Installing project dependencies.") as spinner:
+            result = _wrap_command_with_callbacks(
+                [get_uv_bin(), 'pip', 'install', '--python', '.venv/bin/python', '.'],
                 on_progress=on_progress,
                 on_error=on_error,
             )
+            if result is False:
+                spinner.clear_and_log("Retrying uv installation with --no-cache flag...", 'info')
+                _wrap_command_with_callbacks(
+                    [get_uv_bin(), 'pip', 'install', '--no-cache', '--python', '.venv/bin/python', '.'],
+                    on_progress=on_progress,
+                    on_error=on_error,
+                )
     except Exception as e:
         log.error(f"Installation failed: {str(e)}")
         raise