feat(cli): live hierarchical progress tree and informative error handling

pyproject.toml

@@ -14,8 +14,8 @@ requires-python = ">=3.10,<4.0"
 dependencies = [
     "gevent>=1.5",
     "paramiko>=2.11,<5", # 2.11 (2022) adds Transport.open_channel(timeout=...) for ProxyJump timeout (#971)
-    "click>2",
     "cyclopts>=4,<5",
+    "rich>=13",
     "jinja2>3,<4",
     "python-dateutil>2,<3",
     "typeguard>=4,<5",
@@ -50,7 +50,6 @@ repository = "https://github.com/pyinfra-dev/pyinfra"
 
 [dependency-groups]
 test = [
-    "click>=8.2",
     "pytest>=8.3.5,<9",
     "freezegun>=1.5.5",
     "coverage>=7.7.1,<8",

src/pyinfra/api/connect.py

@@ -24,7 +24,7 @@ def connect_all(state: "State"):
 
     greenlet_to_host = {state.pool.spawn(host.connect): host for host in hosts}
 
-    with progress_spinner(greenlet_to_host.values()) as progress:
+    with progress_spinner(greenlet_to_host.values(), prefix_message="Connecting") as progress:
         for greenlet in gevent.iwait(greenlet_to_host.keys()):
             host = greenlet_to_host[greenlet]
             progress(host)
@@ -57,7 +57,7 @@ def disconnect_all(state: "State"):
         for host in state.activated_hosts  # only hosts we connected to please!
     }
 
-    with progress_spinner(greenlet_to_host.values()) as progress:
+    with progress_spinner(greenlet_to_host.values(), prefix_message="Disconnecting") as progress:
         for greenlet in gevent.iwait(greenlet_to_host.keys()):
             host = greenlet_to_host[greenlet]
             progress(host)

src/pyinfra/api/facts.py

@@ -177,7 +177,12 @@ def get_host_fact(host, *args, **kwargs):
 
     results = {}
 
-    with progress_spinner(greenlet_to_host.values()) as progress:
+    fact_cls = args[0] if args else None
+    fact_name = getattr(fact_cls, "name", None) or getattr(fact_cls, "__name__", "facts")
+
+    with progress_spinner(
+        greenlet_to_host.values(), prefix_message=f"Gathering {fact_name}"
+    ) as progress:
         for greenlet in gevent.iwait(greenlet_to_host.keys()):
             host = greenlet_to_host[greenlet]
             results[host] = greenlet.get()

src/pyinfra/api/host.py

@@ -16,6 +16,7 @@
 
 from pyinfra import logger
 from pyinfra.api.output import format_text
+from pyinfra.api.renderable import Diff, OutputBlock
 from pyinfra.connectors.base import BaseConnector
 from pyinfra.connectors.util import CommandOutput, remove_any_sudo_askpass_file
 from pyinfra.context import ctx_config
@@ -198,19 +199,50 @@ def host_data(self):
     def group_data(self):
         return self.inventory.get_groups_data(self.groups)
 
+    def _styled_name(self, *args, **kwargs) -> str:
+        # Dim any "@connector/" prefix so the host name stands out.
+        name = self.name
+        if name.startswith("@") and "/" in name:
+            connector, _, rest = name.partition("/")
+            return (
+                f"{format_text(f'{connector}/', 'bright_black')}"
+                f"{format_text(rest, *args, **kwargs)}"
+            )
+        return format_text(name, *args, **kwargs)
+
     @property
     def print_prefix(self) -> str:
         if self.nested_executing_op_hash:
-            return f"{format_text('')}[{format_text(self.name, bold=True)}] {format_text('nested', 'blue')}{self.print_prefix_padding} "
+            return f"{self._styled_name('cyan', bold=True)} {format_text('nested', 'blue')}{self.print_prefix_padding} "
 
-        return f"{format_text('')}[{format_text(self.name, bold=True)}]{self.print_prefix_padding} "
+        return f"{self._styled_name('cyan', bold=True)}{self.print_prefix_padding} "
 
     def style_print_prefix(self, *args, **kwargs) -> str:
-        return f"{format_text('')}[{format_text(self.name, *args, **kwargs)}]{self.print_prefix_padding} "
+        return f"{self._styled_name(*args, **kwargs)}{self.print_prefix_padding} "
 
     def log(self, message: str, log_func: Callable[[str], Any] = logger.info) -> None:
         log_func(f"{self.print_prefix}{message}")
 
+    def log_rich(
+        self,
+        descriptor: OutputBlock,
+        log_func: Callable[..., Any] = logger.info,
+    ) -> None:
+        """Emit a rich-free output descriptor as a single log record.
+
+        The descriptor and this host's name are attached to the record via
+        ``extra``; the CLI turns the descriptor into a rich renderable and routes
+        it to this host's tree node (or the console in flat mode). Keeping the
+        descriptor plain lets the core stay decoupled from any rendering library.
+        """
+        log_func("", extra={"pyinfra_rich": descriptor, "pyinfra_host": self.name})
+
+    def log_diff(self, diff_text: str) -> None:
+        """Log a whole (plain) diff block for syntax-highlighted rendering."""
+        if not diff_text:
+            return
+        self.log_rich(Diff(diff_text))
+
     def log_styled(
         self, message: str, log_func: Callable[[str], Any] = logger.info, **kwargs
     ) -> None:

src/pyinfra/api/operations.py

@@ -39,6 +39,7 @@ def run_host_op(state: State, host: Host, op_hash: str) -> bool:
 
     if op_hash not in state.ops[host]:
         logger.info(f"{host.print_prefix}{format_text('Skipped', 'blue')}")
+        state.trigger_callbacks("operation_host_skipped", host, op_hash)
         return True
 
     op_meta = state.get_op_meta(op_hash)
@@ -283,7 +284,7 @@ def _run_serial_ops(state: State):
 
     for host in list(state.inventory.get_active_hosts()):
         host_operations = product([host], state.get_op_order())
-        with progress_spinner(host_operations) as progress:
+        with progress_spinner(host_operations, prefix_message=f"Running ({host.name})") as progress:
             try:
                 _run_host_ops(
                     state,
@@ -300,7 +301,7 @@ def _run_no_wait_ops(state: State):
     """
 
     hosts_operations = product(state.inventory.get_active_hosts(), state.get_op_order())
-    with progress_spinner(hosts_operations) as progress:
+    with progress_spinner(hosts_operations, prefix_message="Running operations") as progress:
         # Spawn greenlet for each host to run *all* ops
         if state.pool is None:
             raise PyinfraError("No pool found on state.")
@@ -326,10 +327,14 @@ def _run_single_op(state: State, op_hash: str):
     op_meta = state.get_op_meta(op_hash)
     log_operation_start(op_meta)
 
+    op_name = ", ".join(op_meta.names) if op_meta.names else "operation"
+
     failed_hosts = set()
 
     if op_meta.global_arguments["_serial"]:
-        with progress_spinner(state.inventory.get_active_hosts()) as progress:
+        with progress_spinner(
+            state.inventory.get_active_hosts(), prefix_message=op_name
+        ) as progress:
             # For each host, run the op
             for host in state.inventory.get_active_hosts():
                 result = _run_host_op_with_context(state, host, op_hash)
@@ -349,7 +354,7 @@ def _run_single_op(state: State, op_hash: str):
             batches = [hosts[i : i + parallel] for i in range(0, len(hosts), parallel)]
 
         for batch in batches:
-            with progress_spinner(batch) as progress:
+            with progress_spinner(batch, prefix_message=op_name) as progress:
                 # Spawn greenlet for each host
                 if state.pool is None:
                     raise PyinfraError("No pool found on state.")
@@ -368,11 +373,13 @@ def _run_single_op(state: State, op_hash: str):
                     if not greenlet.get():
                         failed_hosts.add(host)
 
+    # Signal the operation end first so progress handlers can finalise its
+    # display before fail_hosts potentially prompts or raises.
+    state.trigger_callbacks("operation_end", op_hash)
+
     # Now all the batches/hosts are complete, fail any failures
     state.fail_hosts(failed_hosts)
 
-    state.trigger_callbacks("operation_end", op_hash)
-
 
 def run_ops(state: State, serial: bool = False, no_wait: bool = False):
     """

src/pyinfra/api/output.py

@@ -3,14 +3,50 @@
 
 Provides ``format_text`` and ``echo`` functions that default to plain-text
 no-ops, allowing the API layer to work without any CLI dependency.  The CLI
-layer replaces them at startup via ``set_formatter`` and ``set_echo``.
+layer replaces them at startup via ``set_formatter`` and ``set_echo`` (wiring
+in Rich-backed implementations).
 """
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from collections.abc import Callable
 
+if TYPE_CHECKING:
+    from rich.console import Console
+
+_console: Console | None = None
+
+
+def get_console() -> Console:
+    """
+    Return the shared human-facing (stderr) Rich console.
+
+    Created lazily so importing the API layer doesn't require Rich to be
+    configured.  The CLI may replace it via :func:`set_console` to share a
+    single console between logging, tables and the progress spinner.
+    """
+    global _console
+    if _console is None:
+        from rich.console import Console
+
+        # markup/emoji disabled: host print prefixes like ``[@fake/host]`` and
+        # arbitrary command output must not be interpreted as Rich markup.
+        _console = Console(
+            stderr=True,
+            highlight=False,
+            soft_wrap=True,
+            markup=False,
+            emoji=False,
+        )
+    return _console
+
+
+def set_console(console: Console) -> None:
+    """Replace the shared human-facing console."""
+    global _console
+    _console = console
+
 
 # Default formatter: identity function (returns plain text, ignores styling kwargs).
 def _default_format_text(text: str, *args: Any, **kwargs: Any) -> str:
@@ -29,25 +65,26 @@ def _default_echo(message: Any = None, **kwargs: Any) -> None:
 def format_text(text: str, *args: Any, **kwargs: Any) -> str:
     """Format text with optional styling (color, bold, etc.).
 
-    Mirrors ``click.style`` signature.  Accepts positional ``fg`` argument
-    for compatibility with ``click.style("text", "red")``.
+    Historically mirrored ``click.style``: accepts a positional foreground
+    color (e.g. ``format_text("text", "red")``) and ``bold=`` keyword.  The CLI
+    installs a Rich-backed implementation preserving this signature.
     """
     return _format_text(text, *args, **kwargs)
 
 
 def echo(message: Any = None, **kwargs: Any) -> None:
-    """Echo a message.  Mirrors ``click.echo`` signature."""
+    """Echo a message.  Supports ``err=True`` to write to stderr."""
     return _echo(message, **kwargs)
 
 
 def set_formatter(func: Callable[..., str]) -> None:
-    """Replace the default formatter (e.g. with ``click.style``)."""
+    """Replace the default formatter (e.g. with a Rich-backed styler)."""
     global _format_text
     _format_text = func
 
 
 def set_echo(func: Callable[..., None]) -> None:
-    """Replace the default echo function (e.g. with ``click.echo``)."""
+    """Replace the default echo function (e.g. with a Rich-backed echo)."""
     global _echo
     _echo = func
 

src/pyinfra/api/renderable.py

@@ -0,0 +1,38 @@
+"""
+Rich-free output descriptors for structured CLI output.
+
+Operations and facts describe rich output (diffs, syntax-highlighted code
+blocks, ...) with these plain dataclasses via :meth:`pyinfra.api.host.Host.log_rich`.
+The core stays decoupled from any rendering library — the CLI layer
+(``pyinfra_cli``) maps each descriptor to a Rich renderable via its registry.
+
+New descriptor types only need a subclass here plus a CLI-side renderer
+registration; the routing/rendering pipeline is generic.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class OutputBlock:
+    """Base marker for a structured, rich-free output descriptor."""
+
+
+@dataclass(frozen=True)
+class Diff(OutputBlock):
+    """A unified-diff block (rendered with the ``diff`` lexer)."""
+
+    text: str
+
+
+@dataclass(frozen=True)
+class CodeBlock(OutputBlock):
+    """A syntax-highlighted code block.
+
+    ``lexer`` is a Pygments lexer name (``sql``, ``yaml``, ``json``, ...).
+    """
+
+    text: str
+    lexer: str = "text"

src/pyinfra/api/state.py

@@ -70,6 +70,10 @@ def operation_start(state: State, op_hash):
     def operation_host_start(state: State, host: Host, op_hash):
         pass
 
+    @staticmethod
+    def operation_host_skipped(state: State, host: Host, op_hash):
+        pass
+
     @staticmethod
     def operation_host_success(state: State, host: Host, op_hash, retry_count: int = 0):
         pass

src/pyinfra/api/util.py

@@ -201,7 +201,7 @@ def print_host_combined_output(host: Host, output: CommandOutput) -> None:
 
 
 def log_operation_start(
-    op_meta: StateOperationMeta, op_types: list | None = None, prefix: str = "--> "
+    op_meta: StateOperationMeta, op_types: list | None = None, prefix: str = ""
 ) -> None:
     op_types = op_types or []
     if op_meta.global_arguments["_serial"]:

src/pyinfra/operations/files.py

@@ -66,7 +66,7 @@
     MetadataTimeField,
     adjust_regex,
     ensure_mode_int,
-    generate_color_diff,
+    generate_diff,
     get_timestamp,
     sed_delete,
     sed_replace,
@@ -505,9 +505,7 @@ def line(
     elif present_lines and not present:
         if state.config.DIFF:
             host.log(f"Will Remove lines in {format_text(path, bold=True)}", logger.info)
-            for line in generate_color_diff(present_lines, []):
-                logger.info("  %s", line)
-            logger.info("")
+            host.log_diff("\n".join(generate_diff(present_lines, [])))
         yield sed_delete(
             path,
             match_line,
@@ -525,8 +523,7 @@ def line(
             if state.config.DIFF:
                 host.log(f"Will replace lines in {format_text(path, bold=True)}", logger.info)
                 new_lines = [re.sub(match_line, replace, line) for line in present_lines]
-                for line in generate_color_diff(present_lines, new_lines):
-                    logger.info("  %s", line)
+                host.log_diff("\n".join(generate_diff(present_lines, new_lines)))
             yield sed_replace_command
         else:
             host.noop(f'line "{replace or line}" exists in {path}')
@@ -1159,11 +1156,9 @@ def put(
                 with get_file_io(src, "r") as f:
                     desired_lines = f.readlines()
 
-                for line in generate_color_diff([], desired_lines):
-                    logger.info("  %s", line)
+                host.log_diff("\n".join(generate_diff([], desired_lines)))
             except UnicodeDecodeError:
                 logger.info("Binary file uploaded")
-            logger.info("")
 
         yield FileUploadCommand(
             local_file,
@@ -1208,9 +1203,7 @@ def put(
                 with get_file_io(src, "r") as f:
                     desired_lines = f.readlines()
 
-                for line in generate_color_diff(current_lines, desired_lines):
-                    logger.info("  %s", line)
-                logger.info("")
+                host.log_diff("\n".join(generate_diff(current_lines, desired_lines)))
 
             yield FileUploadCommand(
                 local_file,