feat: enable first-class user feedback (#79)

Author: mtth

poetry.lock

@@ -10,6 +10,7 @@ dependencies = [
   "jinja2 (>=3.1.5,<4)",
   "prettytable (>=3.15.1,<4)",
   "xdg-base-dirs (>=6.0.2,<7)",
+  "yaspin (>=3.1.0,<4)",
 ]
 
 [project.optional-dependencies]

pyproject.toml

@@ -2,22 +2,26 @@
 
 from __future__ import annotations
 
-from collections.abc import Sequence
 import enum
 import importlib.metadata
 import logging
 import optparse
-from pathlib import Path, PurePosixPath
+from pathlib import Path
 import sys
 
 from .bots import load_bot
-from .common import PROGRAM, Config, UnreachableError, ensure_state_home
+from .common import (
+    PROGRAM,
+    Config,
+    Feedback,
+    UnreachableError,
+    ensure_state_home,
+)
 from .drafter import Drafter, DraftMergeStrategy
 from .editor import open_editor
 from .git import Repo
 from .prompt import Template, TemplatedPrompt, find_template, templates_table
 from .store import Store
-from .toolbox import ToolVisitor
 
 
 _logger = logging.getLogger(__name__)
@@ -120,36 +124,6 @@ def merge_strategy(self) -> DraftMergeStrategy | None:
                 raise UnreachableError()
 
 
-class ToolPrinter(ToolVisitor):
-    """Visitor implementation which prints invocations to stdout"""
-
-    def on_list_files(
-        self, _paths: Sequence[PurePosixPath], _reason: str | None
-    ) -> None:
-        print("Listing available files...")
-
-    def on_read_file(
-        self, path: PurePosixPath, _contents: str | None, _reason: str | None
-    ) -> None:
-        print(f"Reading {path}...")
-
-    def on_write_file(
-        self, path: PurePosixPath, _contents: str, _reason: str | None
-    ) -> None:
-        print(f"Wrote {path}.")
-
-    def on_delete_file(self, path: PurePosixPath, _reason: str | None) -> None:
-        print(f"Deleted {path}.")
-
-    def on_rename_file(
-        self,
-        src_path: PurePosixPath,
-        dst_path: PurePosixPath,
-        _reason: str | None,
-    ) -> None:
-        print(f"Renamed {src_path} to {dst_path}.")
-
-
 def edit(*, path: Path | None = None, text: str | None = None) -> str:
     if sys.stdin.isatty():
         return open_editor(text or "", path)
@@ -187,10 +161,11 @@ def main() -> None:  # noqa: PLR0912 PLR0915
         datefmt="%m-%d %H:%M",
     )
 
+    feedback = Feedback.dynamic() if sys.stdin.isatty() else Feedback.static()
     repo = Repo.enclosing(Path(opts.root) if opts.root else Path.cwd())
-    drafter = Drafter.create(repo, Store.persistent())
-    match getattr(opts, "command", "generate"):
-        case "generate":
+    drafter = Drafter.create(repo, Store.persistent(), feedback)
+    match getattr(opts, "command", "new"):
+        case "new":
             bot_config = None
             bot_name = opts.bot or repo.default_bot()
             if bot_name:
@@ -214,29 +189,19 @@ def main() -> None:  # noqa: PLR0912 PLR0915
                 if not prompt or prompt == _PROMPT_PLACEHOLDER:
                     raise ValueError("Aborting: empty or placeholder prompt")
             else:
+                if sys.stdin.isatty():
+                    print("Reading prompt from stdin... (press C-D when done)")
                 prompt = sys.stdin.read()
 
             accept = Accept(opts.accept or 0)
-            draft = drafter.generate_draft(
+            _ = drafter.generate_draft(
                 prompt,
                 bot,
                 prompt_transform=open_editor if editable else None,
                 merge_strategy=accept.merge_strategy(),
-                tool_visitors=[ToolPrinter()],
             )
-            match accept:
-                case Accept.MANUAL:
-                    print(f"Generated draft. [ref={draft.ref}].")
-                case Accept.MERGE | Accept.MERGE_THEIRS:
-                    print(f"Generated and merged draft. [ref={draft.ref}]")
-                case Accept.MERGE_THEN_QUIT:
-                    drafter.quit_folio()
-                    print(f"Generated and applied draft. [ref={draft.ref}]")
-                case _:
-                    raise UnreachableError()
         case "quit":
             drafter.quit_folio()
-            print("Applied draft.")
         case "templates":
             if args:
                 name = args[0]

src/git_draft/__main__.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
+import contextlib
 import dataclasses
 import itertools
 import logging
@@ -15,6 +16,11 @@
 
 import prettytable
 import xdg_base_dirs
+import yaspin
+import yaspin.core
+
+
+_logger = logging.getLogger(__name__)
 
 
 PROGRAM = "git-draft"
@@ -123,3 +129,92 @@ def from_cursor(cls, cursor: sqlite3.Cursor) -> Self:
         table = prettytable.from_db_cursor(cursor, **cls._kwargs)
         assert table
         return cls(table)
+
+
+def _tagged(text: str, /, **kwargs) -> str:
+    tags = [f"{key}={val}" for key, val in kwargs.items() if val is not None]
+    return f"{text} [{', '.join(tags)}]" if tags else text
+
+
+class Feedback:
+    """User feedback interface"""
+
+    def report(self, text: str, **tags) -> None:  # pragma: no cover
+        raise NotImplementedError()
+
+    def spinner(
+        self, text: str, **tags
+    ) -> contextlib.AbstractContextManager[
+        FeedbackSpinner
+    ]:  # pragma: no cover
+        raise NotImplementedError()
+
+    @staticmethod
+    def dynamic() -> Feedback:
+        """Feedback suitable for interactive terminals"""
+        return _DynamicFeedback()
+
+    @staticmethod
+    def static() -> Feedback:
+        """Feedback suitable for pipes, etc."""
+        return _StaticFeedback()
+
+
+class FeedbackSpinner:
+    """Operation feedback tracker"""
+
+    def update(self, text: str, **tags) -> None:  # pragma: no cover
+        raise NotImplementedError()
+
+
+class _DynamicFeedback(Feedback):
+    def __init__(self) -> None:
+        self._spinner: _DynamicFeedbackSpinner | None = None
+
+    def report(self, text: str, **tags) -> None:
+        message = f"☞ {_tagged(text, **tags)}"
+        if self._spinner:
+            self._spinner.yaspin.write(message)
+        else:
+            print(message)  # noqa
+
+    @contextlib.contextmanager
+    def spinner(self, text: str, **tags) -> Iterator[FeedbackSpinner]:
+        assert not self._spinner
+        with yaspin.yaspin(text=_tagged(text, **tags)) as spinner:
+            self._spinner = _DynamicFeedbackSpinner(spinner)
+            try:
+                yield self._spinner
+            except Exception:
+                self._spinner.yaspin.fail("✗")
+                raise
+            else:
+                self._spinner.yaspin.ok("✓")
+            finally:
+                self._spinner = None
+
+
+class _DynamicFeedbackSpinner(FeedbackSpinner):
+    def __init__(self, yaspin: yaspin.core.Yaspin) -> None:
+        self.yaspin = yaspin
+
+    def update(self, text: str, **tags) -> None:
+        self.yaspin.text = _tagged(text, **tags)
+
+
+class _StaticFeedback(Feedback):
+    def report(self, text: str, **tags) -> None:
+        print(_tagged(text, **tags))  # noqa
+
+    @contextlib.contextmanager
+    def spinner(self, text: str, **tags) -> Iterator[FeedbackSpinner]:
+        self.report(text, **tags)
+        yield _StaticFeedbackSpinner(self)
+
+
+class _StaticFeedbackSpinner(FeedbackSpinner):
+    def __init__(self, feedback: _StaticFeedback) -> None:
+        self._feedback = feedback
+
+    def update(self, text: str, **tags) -> None:
+        self._feedback.report(text, **tags)