Add nestable terminal status utility

src/hatch/cli/application.py

@@ -71,30 +71,30 @@ def get_environment(self, env_name=None):
     # used for documenting the life cycle of environments.
     def prepare_environment(self, environment):
         if not environment.exists():
-            with self.status_waiting(f'Creating environment: {environment.name}'):
+            with self.status(f'Creating environment: {environment.name}'):
                 environment.create()
 
             if not environment.skip_install:
                 if environment.pre_install_commands:
-                    with self.status_waiting('Running pre-installation commands'):
+                    with self.status('Running pre-installation commands'):
                         self.run_shell_commands(environment, environment.pre_install_commands, source='pre-install')
 
                 if environment.dev_mode:
-                    with self.status_waiting('Installing project in development mode'):
+                    with self.status('Installing project in development mode'):
                         environment.install_project_dev_mode()
                 else:
-                    with self.status_waiting('Installing project'):
+                    with self.status('Installing project'):
                         environment.install_project()
 
                 if environment.post_install_commands:
-                    with self.status_waiting('Running post-installation commands'):
+                    with self.status('Running post-installation commands'):
                         self.run_shell_commands(environment, environment.post_install_commands, source='post-install')
 
-        with self.status_waiting('Checking dependencies'):
+        with self.status('Checking dependencies'):
             dependencies_in_sync = environment.dependencies_in_sync()
 
         if not dependencies_in_sync:
-            with self.status_waiting('Syncing dependencies'):
+            with self.status('Syncing dependencies'):
                 environment.sync_dependencies()
 
     def run_shell_commands(
@@ -199,7 +199,7 @@ def ensure_plugin_dependencies(self, dependencies: list[Requirement], *, wait_me
         for dependency in dependencies:
             command.append(str(dependency))
 
-        with self.status_waiting(wait_message):
+        with self.status(wait_message):
             self.platform.check_command(command)
 
     def get_env_directory(self, environment_type):
@@ -238,5 +238,6 @@ def __init__(self, app: Application):
         # Divergence from what the backend provides
         self.prompt = app.prompt
         self.confirm = app.confirm
-        self.status_waiting = app.status_waiting
+        self.status = app.status
+        self.status_if = app.status_if
         self.read_builder = app.read_builder

src/hatch/cli/build/__init__.py

@@ -1,5 +1,12 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 import click
 
+if TYPE_CHECKING:
+    from hatch.cli.application import Application
+
 
 @click.command(short_help='Build a project')
 @click.argument('location', required=False)
@@ -43,7 +50,7 @@
 )
 @click.option('--clean-only', is_flag=True, hidden=True)
 @click.pass_obj
-def build(app, location, targets, hooks_only, no_hooks, ext, clean, clean_hooks_after, clean_only):
+def build(app: Application, location, targets, hooks_only, no_hooks, ext, clean, clean_hooks_after, clean_only):
     """Build a project."""
     app.ensure_environment_plugin_dependencies()
 
@@ -101,20 +108,19 @@ def get_version_api(self):
             with environment.get_env_vars(), EnvVars(env_vars):
                 dependencies.extend(builder.config.dependencies)
 
-            with app.status_waiting(
+            with app.status_if(
                 'Setting up build environment', condition=not environment.build_environment_exists()
-            ) as status:
-                with environment.build_environment(dependencies) as build_environment:
-                    status.stop()
-
-                    process = environment.get_build_process(
-                        build_environment,
-                        directory=path,
-                        targets=(target,),
-                        hooks_only=hooks_only,
-                        no_hooks=no_hooks,
-                        clean=clean,
-                        clean_hooks_after=clean_hooks_after,
-                        clean_only=clean_only,
-                    )
-                    app.attach_builder(process)
+            ) as status, environment.build_environment(dependencies) as build_environment:
+                status.stop()
+
+                process = environment.get_build_process(
+                    build_environment,
+                    directory=path,
+                    targets=(target,),
+                    hooks_only=hooks_only,
+                    no_hooks=no_hooks,
+                    clean=clean,
+                    clean_hooks_after=clean_hooks_after,
+                    clean_only=clean_only,
+                )
+                app.attach_builder(process)

src/hatch/cli/env/prune.py

@@ -37,5 +37,5 @@ def prune(app):
             continue
 
         if environment.exists() or environment.build_environment_exists():
-            with app.status_waiting(f'Removing environment: {env_name}'):
+            with app.status(f'Removing environment: {env_name}'):
                 environment.remove()

src/hatch/cli/env/remove.py

@@ -29,5 +29,5 @@ def remove(ctx, env_name):
             continue
 
         if environment.exists() or environment.build_environment_exists():
-            with app.status_waiting(f'Removing environment: {env_name}'):
+            with app.status(f'Removing environment: {env_name}'):
                 environment.remove()

src/hatch/cli/new/__init__.py

@@ -54,7 +54,7 @@ def new(app, name, location, interactive, feature_cli, initialize, setuptools_op
         from hatch.cli.new.migrate import migrate
 
         try:
-            with app.status_waiting('Migrating project metadata from setuptools'):
+            with app.status('Migrating project metadata from setuptools'):
                 migrate(str(location), setuptools_options)
         except Exception as e:
             app.display_error(f'Could not automatically migrate from setuptools: {e}')

src/hatch/cli/project/metadata.py

@@ -35,7 +35,7 @@ def metadata(app, field):
             except Exception as e:
                 app.abort(f'Environment `{environment.name}` is incompatible: {e}')
 
-            with app.status_waiting(
+            with app.status_if(
                 'Setting up build environment for missing dependencies',
                 condition=not environment.build_environment_exists(),
             ) as status, environment.build_environment(app.project.metadata.build.requires):

src/hatch/cli/terminal.py

@@ -1,16 +1,124 @@
 from __future__ import annotations
 
 import os
-from contextlib import contextmanager
+from abc import ABC, abstractmethod
+from functools import cached_property
 from textwrap import indent as indent_text
+from typing import Callable
 
 import click
 from rich.console import Console
 from rich.errors import StyleSyntaxError
+from rich.status import Status
 from rich.style import Style
 from rich.text import Text
 
 
+class TerminalStatus(ABC):
+    @abstractmethod
+    def stop(self) -> None:
+        ...
+
+    def __enter__(self) -> TerminalStatus:
+        return self
+
+    @abstractmethod
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        ...
+
+
+class NullStatus(TerminalStatus):
+    def stop(self):
+        pass
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        pass
+
+
+class BorrowedStatus(TerminalStatus):
+    def __init__(
+        self,
+        console: Console,
+        *,
+        tty: bool,
+        verbosity: int,
+        spinner_style: str,
+        waiting_style: Style,
+        success_style: Style,
+        initializer: Callable,
+        finalizer: Callable,
+    ):
+        self.__console = console
+        self.__tty = tty
+        self.__verbosity = verbosity
+        self.__spinner_style = spinner_style
+        self.__waiting_style = waiting_style
+        self.__success_style = success_style
+        self.__initializer = initializer
+        self.__finalizer = finalizer
+
+        # This is the possibly active current status
+        self.__status: Status | None = None
+
+        # This is used as a stack to display the current message
+        self.__messages: list[tuple[Text, str]] = []
+
+    def stop(self) -> None:
+        if self.__status is not None:
+            self.__status.stop()
+
+    def __call__(self, message: str, final_text: str = '') -> BorrowedStatus:
+        self.__messages.append((Text(message, style=self.__waiting_style), final_text))
+        return self
+
+    def __enter__(self) -> BorrowedStatus:
+        if not self.__messages:
+            return self
+
+        message, _ = self.__messages[-1]
+        if not self.__tty:
+            self.__output(message)
+            return self
+
+        if self.__status is None:
+            self.__initializer()
+        else:
+            self.__status.stop()
+
+        self.__status = self.__console.status(message, spinner=self.__spinner_style)
+        self.__status.start()
+
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        old_message, final_text = self.__messages.pop()
+        if self.__verbosity > 0:
+            if not final_text:
+                final_text = old_message.plain
+                final_text = f'Finished {final_text[:1].lower()}{final_text[1:]}'
+
+            self.__output(Text(final_text, style=self.__success_style))
+
+        if not self.__tty:
+            return
+
+        self.__status.stop()
+        if not self.__messages:
+            self.__status = None
+            self.__finalizer()
+        else:
+            message, _ = self.__messages[-1]
+            self.__status = self.__console.status(message, spinner=self.__spinner_style)
+            self.__status.start()
+
+    def __output(self, text):
+        self.__console.stderr = True
+        try:
+            self.__console.print(text, overflow='ignore', no_wrap=True, crop=False)
+        finally:
+            self.__console.stderr = False
+
+
 class Terminal:
     def __init__(self, verbosity, enable_color, interactive):
         self.verbosity = verbosity
@@ -160,27 +268,21 @@ def display_table(self, title, columns, *, show_lines=False, column_options=None
 
         self.output(table)
 
-    @contextmanager
-    def status_waiting(self, text='', *, final_text=None, condition=True, **kwargs):
-        if not condition or not self.interactive or not self.console.is_terminal:
-            if condition:
-                self.display_waiting(text)
-
-            with MockStatus() as status:
-                yield status
-        else:
-            with self.console.status(Text(text, self._style_level_waiting), spinner=self._style_spinner) as status:
-                try:
-                    self.platform.displaying_status = True
-                    yield status
-
-                    if self.verbosity > 0 and status._live.is_started:
-                        if final_text is None:
-                            final_text = f'Finished {text[:1].lower()}{text[1:]}'
+    @cached_property
+    def status(self) -> BorrowedStatus:
+        return BorrowedStatus(
+            self.console,
+            tty=self.interactive and self.console.is_terminal,
+            verbosity=self.verbosity,
+            spinner_style=self._style_spinner,
+            waiting_style=self._style_level_waiting,
+            success_style=self._style_level_success,
+            initializer=lambda: setattr(self.platform, 'displaying_status', True),  # type: ignore[attr-defined]
+            finalizer=lambda: setattr(self.platform, 'displaying_status', False),  # type: ignore[attr-defined]
+        )
 
-                        self.display_success(final_text)
-                finally:
-                    self.platform.displaying_status = False
+    def status_if(self, *args, condition: bool, **kwargs) -> TerminalStatus:
+        return self.status(*args, **kwargs) if condition else NullStatus()
 
     def output(self, text='', style=None, *, stderr=False, indent=None, link=None, **kwargs):
         kwargs.setdefault('overflow', 'ignore')
@@ -213,14 +315,3 @@ def prompt(text, **kwargs):
     @staticmethod
     def confirm(text, **kwargs):
         return click.confirm(text, **kwargs)
-
-
-class MockStatus:
-    def stop(self):
-        pass
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        pass

src/hatch/cli/version/__init__.py

@@ -44,7 +44,7 @@ def version(app, desired_version):
             except Exception as e:
                 app.abort(f'Environment `{environment.name}` is incompatible: {e}')
 
-            with app.status_waiting(
+            with app.status_if(
                 'Setting up build environment for missing dependencies',
                 condition=not environment.build_environment_exists(),
             ) as status, environment.build_environment(app.project.metadata.build.requires):