feat: implement a preable writer to help with logging

src/cipy/action.py

@@ -4,6 +4,7 @@ import pathlib
 import subprocess
 import tempfile
 
+from textwrap import dedent
 from typing import Any, final
 
 from pydantic import Field, PrivateAttr
@@ -102,18 +103,20 @@ class Script(Action):
     """Action descriptor for a generic shell runner"""
 
     shell: Shell = Shell.DEFAULT
+    name: str = ""
     script: str
     _lines: list[str] = PrivateAttr()
 
     def model_post_init(self, context: Any, /) -> None:
+        self.script = dedent(self.script).strip("\n")
         self._lines = self.script.splitlines()
-        if self.name is None:
+        if not self.name:
             self.name = self._lines[0]
 
     @final
     @cipy.runner.ipc
+    @cipy.runner.preamble(extra=[("CYAN", lambda s: s._lines)])
     def run(self, context: Context) -> Status:
-        self.logger.info("Running %s script:\n%s", self.shell, self.script)
         with tempfile.TemporaryFile(mode="w+", suffix=self.shell.extension()) as script:
             script.write(self.script)
             try:

src/cipy/runner.py

@@ -7,14 +7,19 @@ import os
 import tempfile
 
 from contextlib import contextmanager
-from typing import Any, Callable, Iterator, TypeVar
+from typing import Any, Callable, Iterable, Iterator, Literal, TypeVar, overload
 
 from dotenv import dotenv_values
 
 import cipy.common
 from cipy.common import Context, Status, _validate
 
+import cipy._logging as _logging
+
 Action = TypeVar("Action", bound=cipy.common.Action)
+type Run[Action] = Callable[[Action, Context], Status]
+type GetLines[Action] = Callable[[Action], Iterable[str]]
+type ExtraFormatting[Action] = list[tuple[str, GetLines[Action]]]
 
 
 @contextmanager
@@ -37,9 +42,7 @@ def environ(*, error_on_override: bool = True, **overrides: Any) -> Iterator[Non
             os.environ[key] = value
 
 
-def ipc(
-    func: Callable[[Action, Context], Status],
-) -> Callable[[Action, Context], Status]:
+def ipc(func: Run[Action]) -> Run[Action]:
     """
     IPC tool for passing inputs and outputs between an Action that is
     implemented as some manner of script.
@@ -73,3 +76,59 @@ def ipc(
         return rval
 
     return wrapper
+
+
+@overload
+def preamble(
+    func: Run[Action],
+    *,
+    extra: ExtraFormatting[Action] | None = None,
+) -> Run[Action]: ...
+
+
+@overload
+def preamble(
+    func: ExtraFormatting[Action],
+    *,
+    extra: ExtraFormatting[Action] | None = None,
+) -> Callable[
+    [Run[Action]],
+    Run[Action],
+]: ...
+
+
+@overload
+def preamble(
+    func: None = None, *, extra: ExtraFormatting[Action] | None = None
+) -> Callable[[Run[Action]], Run[Action]]: ...
+
+
+def preamble(
+    func: Run[Action] | ExtraFormatting[Action] | None = None,
+    *,
+    extra: ExtraFormatting[Action] | None = None,
+):
+    if isinstance(func, list):
+        assert extra is None
+        return lambda f: preamble(f, extra=func)
+
+    if func is None:
+        assert extra is not None
+        return lambda f: preamble(f, extra=extra)
+
+    if extra is None:
+        extra = []
+
+    @functools.wraps(func)
+    def wrapper(self: Action, context: Context) -> Status:
+        self.logger.info("##[group] Run %s", self.name)
+        for color, get_lines in extra:
+            fmt = "%s"
+            if color in _logging.__dict__:
+                fmt = f"{_logging.__dict__[color]}%s{_logging.RESET}"
+            for line in get_lines(self):
+                self.logger.info(fmt, line)
+        self.logger.info("##[endgroup]")
+        return func(self, context)
+
+    return wrapper