Переглянути джерело

feat: implement mustache templates for Script, fix file writing

Sam Jaffe 1 місяць тому
батько
коміт
50e6c4ffbb
3 змінених файлів з 36 додано та 7 видалено
  1. 1 0
      pyproject.toml
  2. 19 0
      src/cipy/_io.py
  3. 16 7
      src/cipy/action.py

+ 1 - 0
pyproject.toml

@@ -5,6 +5,7 @@ description = "Python based tool for operating CI Pipelines"
 readme = "README.md"
 requires-python = ">=3.14"
 dependencies = [
+    "chevron>=0.14.0",
     "colored>=2.3.2",
     "pydantic>=2.12.5",
     "python-dotenv>=1.2.2",

+ 19 - 0
src/cipy/_io.py

@@ -0,0 +1,19 @@
+"""Utility for manipulating I/O streams"""
+
+import contextlib
+import io
+
+@contextlib.contextmanager
+def capture_stdout():
+    wrap = io.StringIO()
+    with contextlib.redirect_stdout(wrap):
+        yield wrap
+    wrap.close()
+
+
+@contextlib.contextmanager
+def capture_stderr():
+    wrap = io.StringIO()
+    with contextlib.redirect_stderr(wrap):
+        yield wrap
+    wrap.close()

+ 16 - 7
src/cipy/action.py

@@ -7,6 +7,7 @@ import tempfile
 from textwrap import dedent
 from typing import Any, ClassVar, final
 
+from chevron import render  # type: ignore[import-untyped]
 from colored import Fore, Style
 from pydantic import Field, PrivateAttr
 
@@ -15,6 +16,8 @@ from cipy import settings
 from cipy.common import Action, Context, Factory, Ref, Results, Status, _validate
 from cipy.shell import Shell
 
+from . import _io
+
 
 class Call(Action, extra="allow"):
     """
@@ -111,28 +114,34 @@ class Script(Action):
     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 not self.name:
             self.name = f"a {self.shell} script"
 
     @final
     @cipy.runner.ipc
     def run(self, context: Context) -> Status:
-        with cipy.runner.logging_group(self, self._lines[0], ci_only=True):
+        with _io.capture_stderr() as stderr:
+            script = render(self.script, context, warn=True)
+            if stderr.getvalue():
+                self.logger.error(stderr.getvalue())
+                return Status.FAILURE
+            lines = script.splitlines()
+
+        with cipy.runner.logging_group(self, lines[0], ci_only=True):
             if settings.INTERACTIVE:
                 self.logger.warning("Run %s", self.name)
-            for line in self._lines:
+            for line in lines:
                 self.logger.info("%s%s%s", Script._GREEN, line, Style.reset)
             self.logger.info("shell: %s", " ".join(self.shell.command("{0}")))
 
-        with tempfile.NamedTemporaryFile(suffix=self.shell.extension()) as script:
-            script.write(self.script.encode("utf-8"))
+        with tempfile.NamedTemporaryFile(suffix=self.shell.extension()) as file:
+            file.write(script.encode("utf-8"))
+            file.flush()
             try:
-                subprocess.run(self.shell.command(script.name), check=True)
+                subprocess.run(self.shell.command(file.name), check=True)
                 return Status.SUCCESS
             except subprocess.CalledProcessError:
                 return Status.FAILURE