Enable auto extension detection

kapitan/cli.py

@@ -20,7 +20,7 @@
 
 from kapitan import cached, defaults, setup_logging
 from kapitan.initialiser import initialise_skeleton
-from kapitan.inputs.jsonnet import jsonnet_file
+from kapitan.inputs.jsonnet import select_jsonnet_runtime
 from kapitan.inventory import AVAILABLE_BACKENDS, InventoryBackends
 from kapitan.lint import start_lint
 from kapitan.refs.base import RefController, Revealer
@@ -49,7 +49,7 @@ def trigger_eval(args):
     def _search_imports(cwd, imp):
         return search_imports(cwd, imp, search_paths)
 
-    json_output = jsonnet_file(
+    json_output = select_jsonnet_runtime(
         file_path,
         import_callback=_search_imports,
         native_callbacks=resource_callbacks(search_paths),

kapitan/inputs/base.py

@@ -18,9 +18,9 @@
 
 from kapitan import cached
 from kapitan.errors import CompileError, KapitanError
-from kapitan.inventory.model.input_types import CompileInputTypeConfig
+from kapitan.inventory.model.input_types import CompileInputTypeConfig, OutputType
 from kapitan.refs.base import Revealer
-from kapitan.utils import PrettyDumper
+from kapitan.utils import PrettyDumper, prune_empty
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +35,7 @@ class InputType(object):
     """
 
     __metaclass__ = abc.ABCMeta
+    output_type_default = OutputType.YAML
 
     def __init__(self, compile_path, search_paths, ref_controller, target_name, args):
         self.compile_path = compile_path
@@ -72,13 +73,80 @@ def compile_obj(self, comp_obj: CompileInputTypeConfig):
         for expanded_path in expanded_input_paths:
             self.compile_input_path(comp_obj, expanded_path)
 
+    def to_file(self, config: CompileInputTypeConfig, file_path, file_content):
+        """Write compiled content to file, handling different output types and revealing refs if needed.
+
+        Args:
+            config: CompileInputTypeConfig object.
+            file_path: Path to the output file.
+            file_content: Compiled content to write.
+
+        Raises:
+            ValueError: If the output type is not supported.
+        """
+        reveal = self.args.reveal
+        target_name = self.target_name
+
+        indent = self.args.indent
+        output_type = config.output_type
+        file_ext = output_type
+
+        if config.prune:
+            file_content = prune_empty(file_content)
+
+        if output_type == OutputType.AUTO:
+            _, detected_type = os.path.splitext(file_path)
+            if detected_type:
+                # Remove . from the beginning of the extension
+                detected_type = detected_type[1:]
+
+            if detected_type in [OutputType.TOML, OutputType.JSON, OutputType.YAML, OutputType.YML]:
+                output_type = detected_type
+                file_ext = None
+            else:
+                # Extension is not handled, falling back to input type default
+                output_type = self.output_type_default
+                logger.debug("Could not detect extension for %s, defaulting to %s", file_path, output_type)
+                file_ext = output_type  # no extension for plain text
+
+        if output_type == OutputType.PLAIN:
+            file_ext = None  # no extension for plain text
+
+        file_name = f"{file_path}.{file_ext}" if file_ext else file_path
+
+        with CompiledFile(
+            # file_path: path to the output file
+            file_name,
+            # ref_controller: reference controller to resolve refs
+            self.ref_controller,
+            # mode: file open mode, 'w' for write
+            mode="w",
+            # reveal: reveal refs in output
+            reveal=reveal,
+            target_name=target_name,
+            indent=indent,
+        ) as fp:
+            if output_type == OutputType.JSON:
+                fp.write_json(file_content)
+            elif output_type in [OutputType.YAML, OutputType.YML]:
+                fp.write_yaml(file_content)
+            elif output_type == OutputType.PLAIN:
+                fp.write(file_content)
+            elif output_type == OutputType.TOML:
+                fp.write_toml(file_content)
+            else:
+                raise ValueError(
+                    f"Output type defined in inventory for {config} not supported: {output_type}: {OutputType}"
+                )
+
     def compile_input_path(self, comp_obj: CompileInputTypeConfig, input_path: str):
-        """Compile a single input path.
+        """Compile a single input path and write the result to the output directory.
+
+        Creates the output directory if it doesn't exist.
 
         Args:
             comp_obj: CompileInputTypeConfig object.
             input_path: Path to the input file.
-
         Raises:
             CompileError: If compilation fails.
 
@@ -87,6 +155,7 @@ def compile_input_path(self, comp_obj: CompileInputTypeConfig, input_path: str):
         output_path = comp_obj.output_path
 
         logger.debug("Compiling %s", input_path)
+
         try:
             target_compile_path = os.path.join(self.compile_path, target_name.replace(".", "/"), output_path)
             os.makedirs(target_compile_path, exist_ok=True)
@@ -98,33 +167,52 @@ def compile_input_path(self, comp_obj: CompileInputTypeConfig, input_path: str):
             )
 
         except KapitanError as e:
-            raise CompileError("{}\nCompile error: failed to compile target: {}".format(e, target_name))
+            raise CompileError(
+                "{}\nCompile error: failed to compile target: {}".format(e, target_name)
+            ) from e
 
     @abc.abstractmethod
     def compile_file(self, config: CompileInputTypeConfig, input_path: str, compile_path: str):
         """Compile a single input file.
 
+        This is an abstract method that should be implemented by subclasses to handle
+        specific input formats.
+
         Args:
             config: CompileInputTypeConfig object.
             input_path: Path to the input file.
             compile_path: Path to the output directory.
-
-        Raises:
-            NotImplementedError: This is an abstract method.
-
         """
         return NotImplementedError
 
 
 class CompilingFile(object):
-    def __init__(self, context, fp, ref_controller, **kwargs):
+    """
+    A class to handle writing compiled data to a file.
+
+    Provides methods to write data in different formats (YAML, JSON, TOML) to a file,
+    handling reference revealing and compilation as needed.
+
+    Args:
+        fp: File object to write to.
+        ref_controller: Reference controller to resolve references.
+        **kwargs: Additional keyword arguments (e.g., reveal, target_name, indent).
+    """
+
+    def __init__(self, fp, ref_controller, **kwargs):
         self.fp = fp
-        self.ref_controller = ref_controller
         self.kwargs = kwargs
         self.revealer = Revealer(ref_controller)
 
     def write(self, data):
-        """write data into file"""
+        """Write data into file.
+
+        Args:
+            data: Data to write.
+
+        Reveals references if `reveal` is True in kwargs.
+
+        """
         reveal = self.kwargs.get("reveal", False)
         target_name = self.kwargs.get("target_name", None)
 
@@ -133,8 +221,14 @@ def write(self, data):
         else:
             self.fp.write(self.revealer.compile_raw(data, target_name=target_name))
 
-    def write_yaml(self, obj):
-        """recursively compile or reveal refs and convert obj to yaml and write to file"""
+    def write_yaml(self, obj: object):
+        """Recursively compile or reveal refs and convert obj to YAML and write to file.
+
+        Args:
+            obj: Object to write.
+
+        Uses PrettyDumper to handle multiline strings according to style selection.
+        """
         indent = self.kwargs.get("indent", 2)
         reveal = self.kwargs.get("reveal", False)
         target_name = self.kwargs.get("target_name", None)
@@ -177,23 +271,36 @@ def write_yaml(self, obj):
         else:
             logger.debug("%s is Empty, skipped writing output", self.fp.name)
 
-    def write_json(self, obj):
-        """recursively hash or reveal refs and convert obj to json and write to file"""
+    def write_json(self, obj: object):
+        """Recursively compile or reveal refs and convert obj to JSON and write to file.
+
+        Args:
+            obj: Object to write.
+
+        Reveals references if `reveal` is True in kwargs.
+        """
         indent = self.kwargs.get("indent", 2)
         reveal = self.kwargs.get("reveal", False)
         target_name = self.kwargs.get("target_name", None)
         if reveal:
             obj = self.revealer.reveal_obj(obj)
         else:
             obj = self.revealer.compile_obj(obj, target_name=target_name)
+
         if obj:
             json.dump(obj, self.fp, indent=indent)
             logger.debug("Wrote %s", self.fp.name)
         else:
             logger.debug("%s is Empty, skipped writing output", self.fp.name)
 
-    def write_toml(self, obj):
-        """recursively compile or reveal refs and convert obj to toml and write to file"""
+    def write_toml(self, obj: object):
+        """Recursively compile or reveal refs and convert obj to TOML and write to file.
+
+        Args:
+            obj: Object to write.
+
+        Reveals references if `reveal` is True in kwargs.
+        """
         reveal = self.kwargs.get("reveal", False)
         target_name = self.kwargs.get("target_name", None)
         if reveal:
@@ -208,6 +315,16 @@ def write_toml(self, obj):
 
 
 class CompiledFile(object):
+    """
+    A context manager to handle writing compiled data to a file.
+
+    Args:
+        name: Name of the output file.
+        ref_controller: Reference controller to resolve references.
+        **kwargs: Additional keyword arguments (e.g., mode).
+
+    """
+
     def __init__(self, name, ref_controller, **kwargs):
         self.name = name
         self.fp = None
@@ -221,7 +338,7 @@ def __enter__(self):
         os.makedirs(os.path.dirname(self.name), exist_ok=True)
 
         self.fp = open(self.name, mode)
-        return CompilingFile(self, self.fp, self.ref_controller, **self.kwargs)
+        return CompilingFile(self.fp, self.ref_controller, **self.kwargs)
 
     def __exit__(self, *args):
         self.fp.close()