Add create-spec cli pipe support

commands/create_spec.py

@@ -7,36 +7,56 @@
 from prompt import collect_var_value
 
 
-async def handle_create_spec(spec_template: str, output_file: str | None = None, var_overrides: list[str] | None = None, verbose: bool = False):
-    """Handle the create-spec command"""
-
-    # Configure logging based on verbose flag
+async def handle_create_spec(
+    spec_template: str | None,
+    output_file: str | None = None,
+    var_overrides: list[str] | None = None,
+    verbose: bool = False,
+):
     log_level = logging.DEBUG if verbose else logging.WARNING
-    logging.basicConfig(level=log_level, format='%(message)s', stream=sys.stdout, force=True)
+    logging.basicConfig(level=log_level, format="%(message)s", force=True)
+
+    # Detect piped input (stdin not a TTY) and ensure there's data before using it
+    stdin_piped = not sys.stdin.isatty()
+
+    # Determine template source
+    template_engine: TemplateEngine
+    if spec_template:
+        # Resolve relative paths against current working directory
+        template_path = os.path.abspath(spec_template)
 
-    # Resolve relative paths against current working directory
-    template_path = os.path.abspath(spec_template)
+        # Check if template file exists
+        if not os.path.exists(template_path):
+            logging.error(f"Error: Template file '{spec_template}' not found")
+            sys.exit(1)
 
-    # Check if template file exists
-    if not os.path.exists(template_path):
-        print(f"Error: Template file '{spec_template}' not found", file=sys.stderr)
+        template_engine = TemplateEngine.from_file(template_path)
+    elif stdin_piped:
+        try:
+            template_str = sys.stdin.read()
+        except Exception as e:
+            logging.error(f"Error: Failed to read from stdin: {e}")
+            sys.exit(1)
+        if not template_str:
+            logging.error("Error: No data received on stdin for template")
+            sys.exit(1)
+
+        template_engine = TemplateEngine.from_string(template_str)
+    else:
+        logging.error("Error: Missing spec_template argument (or provide template via stdin)")
         sys.exit(1)
 
     try:
-        # Initialize template engine
-        template_engine = TemplateEngine(template_path)
-
-        # Get variables from template
         variables = template_engine.get_variables()
 
         # Parse var_overrides into a dictionary
         provided_vars = {}
         if var_overrides:
             for var_override in var_overrides:
-                if '=' not in var_override:
-                    print(f"Error: Invalid variable format '{var_override}'. Use KEY=VALUE format.", file=sys.stderr)
+                if "=" not in var_override:
+                    logging.error(f"Error: Invalid variable format '{var_override}'. Use KEY=VALUE format.")
                     sys.exit(1)
-                key, value = var_override.split('=', 1)
+                key, value = var_override.split("=", 1)
                 provided_vars[key] = value
 
         # Collect values for each variable
@@ -46,13 +66,13 @@ async def handle_create_spec(spec_template: str, output_file: str | None = None,
             for var in sorted(variables):
                 if var in provided_vars:
                     raw_value = provided_vars[var]
-                    logging.info(f"  {var}: {raw_value}")
+
                 else:
                     raw_value = await collect_var_value(var)
-                    logging.info(f"  {var}: {raw_value}")
+                logging.info(f"  {var}: {raw_value}")
 
                 # Try to parse as JSON if it looks like JSON
-                if raw_value and (raw_value.strip().startswith('{') or raw_value.strip().startswith('[')):
+                if raw_value and (raw_value.strip().startswith("{") or raw_value.strip().startswith("[")):
                     try:
                         collected_vars[var] = json.loads(raw_value)
                     except json.JSONDecodeError:
@@ -69,16 +89,16 @@ async def handle_create_spec(spec_template: str, output_file: str | None = None,
         # Output to file or stdout
         if output_file:
             output_path = os.path.abspath(output_file)
-            with open(output_path, 'w') as f:
+            with open(output_path, "w") as f:
                 f.write(rendered_content)
             logging.info(f"Rendered template saved to: {output_path}")
         else:
             logging.debug("\nRendered template:")
             print(rendered_content)
 
     except TemplateParseError as e:
-        print(f"Error: {e}", file=sys.stderr)
+        logging.error(f"Error: {e}")
         sys.exit(1)
     except Exception as e:
-        print(f"Error: Failed to process template: {e}", file=sys.stderr)
+        logging.error(f"Error: Failed to process template: {e}")
         sys.exit(1)

cxk.py

@@ -23,7 +23,7 @@ async def main():
 
     # cxk create-spec [spec-template]
     create_spec_parser = subparsers.add_parser("create-spec", help="Create spec from template")
-    create_spec_parser.add_argument("spec_template", help="Path to the spec template file")
+    create_spec_parser.add_argument("spec_template", nargs=argparse.OPTIONAL, help="Path to the spec template file")
     create_spec_parser.add_argument("--output", help="Output file path (defaults to stdout if not specified)")
     create_spec_parser.add_argument("--var", action="append", help="Set template variable value (format: key=value)")
     create_spec_parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose output")

engine/__init__.py

@@ -1,36 +1,126 @@
-import os
+from pathlib import Path
 from typing import Any
 
-from jinja2 import Environment, FileSystemLoader, meta, select_autoescape
+from jinja2 import Environment, FileSystemLoader, Template, meta, select_autoescape
 
 
 class TemplateParseError(Exception):
+    """Raised when template parsing fails"""
+
     pass
 
 
 class TemplateEngine:
-    """Abstract away the jinja2 template engine"""
+    """Abstract away the jinja2 template engine with clean factory methods"""
+
+    def __init__(
+        self, env: Environment, template: Template, source_path: Path | None = None, source_string: str | None = None
+    ):
+        """Private constructor - use from_file() or from_string() instead"""
+        self.env = env
+        self.template = template
+        self._source_path = source_path
+        self._source_string = source_string
+
+    @classmethod
+    def from_file(cls, path: str | Path) -> "TemplateEngine":
+        """Create a TemplateEngine from a template file.
+
+        Args:
+            path: Path to the template file
+
+        Returns:
+            TemplateEngine instance
+
+        Raises:
+            FileNotFoundError: If template file doesn't exist
+            TemplateParseError: If template parsing fails
+        """
+        path = Path(path)
+
+        if not path.exists():
+            raise FileNotFoundError(f"Template file not found: {path}")
+
+        template_dir = path.parent
+        template_name = path.name
+
+        env = Environment(
+            loader=FileSystemLoader(str(template_dir)),
+            autoescape=select_autoescape(),
+            enable_async=True,
+        )
+
+        try:
+            template = env.get_template(template_name)
+        except Exception as e:
+            raise TemplateParseError(f"Failed to load template from {path}: {e}") from e
 
-    def __init__(self, template_path: str):
-        template_dir = os.path.dirname(template_path)
-        template_name = os.path.basename(template_path)
+        return cls(env=env, template=template, source_path=path, source_string=None)
 
-        self.env = Environment(
-            loader=FileSystemLoader(template_dir),
+    @classmethod
+    def from_string(cls, template_string: str, name: str = "<stdio>") -> "TemplateEngine":
+        """Create a TemplateEngine from a template string.
+
+        Args:
+            template_string: The template content as a string
+            name: Optional name for the template (for debugging)
+
+        Returns:
+            TemplateEngine instance
+
+        Raises:
+            TemplateParseError: If template parsing fails
+        """
+
+        env = Environment(
             autoescape=select_autoescape(),
             enable_async=True,
         )
-        self.template = self.env.get_template(template_name)
-        self.template_path = template_path
+
+        try:
+            template = env.from_string(template_string)
+        except Exception as e:
+            raise TemplateParseError(f"Failed to parse template string: {e}") from e
+
+        # Store the name in the template for better error messages
+        template.name = name
+
+        return cls(env=env, template=template, source_path=None, source_string=template_string)
+
+    @property
+    def source(self) -> str:
+        """Get the template source content."""
+        if self._source_string is not None:
+            return self._source_string
+
+        if self._source_path is not None:
+            with open(self._source_path, encoding="utf-8") as f:
+                return f.read()
+
+        # Should not reach here with proper factory method usage
+        raise AssertionError("No template source available")
+
+    @property
+    def path(self) -> Path | None:
+        """Get the template file path if loaded from file."""
+        return self._source_path
+
+    @property
+    def is_from_file(self) -> bool:
+        """Check if template was loaded from a file."""
+        return self._source_path is not None
 
     def get_variables(self) -> set[str]:
-        """Get the free variables in the template"""
+        """Get the free (undeclared) variables in the template.
 
-        with open(self.template_path, encoding="utf-8") as f:
-            template_source = f.read()
+        Returns:
+            Set of variable names that are referenced but not defined in the template
 
+        Raises:
+            TemplateParseError: If template parsing fails
+        """
         try:
-            ast = self.env.parse(template_source)
+            ast = self.env.parse(self.source)
         except Exception as e:
             raise TemplateParseError(f"Failed to parse template: {e}") from e
 
@@ -40,3 +130,10 @@ def get_variables(self) -> set[str]:
 
     async def render_async(self, *args: Any, **kwargs: Any) -> str:
         return await self.template.render_async(*args, **kwargs)
+
+    def __repr__(self) -> str:
+        if self._source_path:
+            return f"TemplateEngine(from_file={self._source_path})"
+        else:
+            name = self.template.name if hasattr(self.template, "name") else "<stdio>"
+            return f"TemplateEngine(from_string, name={name})"

tests/README.md

@@ -22,4 +22,9 @@ uv run cxk.py create-spec tests/templates/spec1.md --var additional_context=aa -
 ### With verbose, vars and output
 ```
 uv run cxk.py create-spec tests/templates/spec1.md --verbose --var additional_context=1 --var ticket='{"id":1}' --output res.md
+```
+
+### Piped
+```
+cat tests/templates/spec1.md | uv run cxk.py create-spec --verbose --var ticket='{"id":1}' --var additional_context=2
 ```

tests/e2e/test_e2e.py

@@ -297,12 +297,6 @@ def test_create_spec_with_variables(self, temp_non_git_dir):
         result = self.run_cli(["create-spec", "--verbose", str(template_file)], use_test_runner=True)
 
         assert result.returncode == 0
-        assert "Collecting values for template variables:" in result.stdout
-        # Verify mocked values are displayed
-        assert "age: 25" in result.stdout
-        assert "city: New York" in result.stdout
-        assert "name: John" in result.stdout
-        assert 'weather: {"condition": "sunny", "temp": "75F"}' in result.stdout
 
         # Verify rendered template output
         assert "Hello John!" in result.stdout
@@ -320,10 +314,9 @@ def test_create_spec_no_variables(self, temp_non_git_dir):
         result = self.run_cli(["create-spec", "--verbose", str(template_file)], use_test_runner=True)
 
         assert result.returncode == 0
-        assert "No variables found in template" in result.stdout
+        assert "No variables found in template" in result.stderr
 
         # Verify rendered template output for static template
-        assert "Rendered template:" in result.stdout
         assert "This is a static template with no variables." in result.stdout
 
     def test_create_spec_relative_path(self, temp_non_git_dir):
@@ -339,8 +332,7 @@ def test_create_spec_relative_path(self, temp_non_git_dir):
         )
 
         assert result.returncode == 0
-        assert "Collecting values for template variables:" in result.stdout
-        assert "username: testuser" in result.stdout
+        assert "username: testuser" in result.stderr
 
         # Verify rendered template output
         assert "Hello testuser!" in result.stdout
@@ -597,7 +589,7 @@ def test_create_spec_with_var_and_output_file(self, temp_non_git_dir):
         )
 
         assert result.returncode == 0
-        assert f"Rendered template saved to: {output_file}" in result.stdout
+        assert f"Rendered template saved to: {output_file}" in result.stderr
 
         # Verify file was created and contains expected content
         assert output_file.exists()