Refactor input types (#1272)

Fixes #

## Proposed Changes

* Refactor input types code to make it easier to manage and extend
* Allow dynamic import of input types

## Docs and Tests

* [x] Tests added
* [x] Updated documentation

kapitan/cli.py

@@ -74,27 +74,10 @@ def trigger_compile(args):
     cached.revealer_obj = Revealer(ref_controller)
 
     compile_targets(
-        args.inventory_path,
-        search_paths,
-        args.output_path,
-        args.parallelism,
-        args.targets,
-        args.labels,
-        ref_controller,
-        prune=args.prune,
-        indent=args.indent,
-        reveal=args.reveal,
-        cache=args.cache,
-        cache_paths=args.cache_paths,
-        fetch=args.fetch,
-        force_fetch=args.force_fetch,
-        force=args.force,  # deprecated
-        validate=args.validate,
-        schemas_path=args.schemas_path,
-        jinja2_filters=args.jinja2_filters,
-        verbose=args.verbose,
-        use_go_jsonnet=args.use_go_jsonnet,
-        compose_target_name=args.compose_target_name,
+        inventory_path=args.inventory_path,
+        search_paths=search_paths,
+        ref_controller=ref_controller,
+        args=args,
     )
 
 

kapitan/inputs/__init__.py

@@ -1,3 +1,36 @@
 # SPDX-FileCopyrightText: 2020 The Kapitan Authors <[email protected]>
 #
 # SPDX-License-Identifier: Apache-2.0
+
+import functools  # Import functools for caching
+import importlib
+from typing import Type
+
+from kapitan.inventory.model.input_types import InputTypes
+
+from .base import InputType
+
+
+@functools.lru_cache(maxsize=None)  # Use lru_cache for caching
+def get_compiler(input_type: InputType) -> Type[InputType]:
+    """Dynamically imports and returns the compiler class based on input_type."""
+
+    module_map = {
+        InputTypes.JINJA2: "jinja2",
+        InputTypes.HELM: "helm",
+        InputTypes.JSONNET: "jsonnet",
+        InputTypes.KADET: "kadet",
+        InputTypes.COPY: "copy",
+        InputTypes.EXTERNAL: "external",
+        InputTypes.REMOVE: "remove",
+    }
+
+    module_name = module_map.get(input_type)
+    if module_name:
+        try:
+            module = importlib.import_module(f".{module_name}", package=__name__)
+            return getattr(module, module_name.capitalize())  # Capitalize to get class name
+        except (ImportError, AttributeError) as e:
+            raise ImportError(f"Could not import module or class for {input_type}: {e}") from e
+    else:
+        return None  # Or raise an appropriate error for unknown input_type

kapitan/inputs/base.py

@@ -5,6 +5,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import abc
 import glob
 import itertools
 import json
@@ -17,30 +18,44 @@
 
 from kapitan import cached
 from kapitan.errors import CompileError, KapitanError
+from kapitan.inventory.model.input_types import CompileInputTypeConfig
 from kapitan.refs.base import Revealer
 from kapitan.utils import PrettyDumper
 
 logger = logging.getLogger(__name__)
 
 
 class InputType(object):
-    def __init__(self, type_name, compile_path, search_paths, ref_controller):
-        self.type_name = type_name
+    """
+    Abstract base class for input types.
+
+    Provides methods for compiling input files.  Subclasses should implement
+    the `compile_file` method to handle specific input formats.
+
+    """
+
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, compile_path, search_paths, ref_controller, target_name, args):
         self.compile_path = compile_path
         self.search_paths = search_paths
         self.ref_controller = ref_controller
+        self.target_name = target_name
+        self.args = args
+
+    def compile_obj(self, comp_obj: CompileInputTypeConfig):
+        """Expand globbed input paths and compile each resolved input path.
+
+        Args:
+            comp_obj: CompileInputTypeConfig object containing input paths and other compilation options.
+
+        Raises:
+            CompileError: If an input path is not found and ignore_missing is False.
 
-    def compile_obj(self, comp_obj, ext_vars, **kwargs):
-        """
-        Expand globbed input paths, taking into account provided search paths
-        and run compile_input_path() for each resolved input_path.
-        kwargs are passed into compile_input_path()
         """
-        input_type = comp_obj.input_type
-        assert input_type == self.type_name
 
         # expand any globbed paths, taking into account provided search paths
-        input_paths = []
+        expanded_input_paths = []
         for input_path in comp_obj.input_paths:
             globbed_paths = [glob.glob(os.path.join(path, input_path)) for path in self.search_paths]
             inputs = list(itertools.chain.from_iterable(globbed_paths))
@@ -50,57 +65,54 @@ def compile_obj(self, comp_obj, ext_vars, **kwargs):
             if len(inputs) == 0 and not ignore_missing:
                 raise CompileError(
                     "Compile error: {} for target: {} not found in "
-                    "search_paths: {}".format(input_path, ext_vars["target"], self.search_paths)
+                    "search_paths: {}".format(input_path, self.target_name, self.search_paths)
                 )
-            input_paths.extend(inputs)
+            expanded_input_paths.extend(inputs)
 
-        for input_path in input_paths:
-            self.compile_input_path(input_path, comp_obj, ext_vars, **kwargs)
+        for expanded_path in expanded_input_paths:
+            self.compile_input_path(comp_obj, expanded_path)
+
+    def compile_input_path(self, comp_obj: CompileInputTypeConfig, input_path: str):
+        """Compile a single input path.
+
+        Args:
+            comp_obj: CompileInputTypeConfig object.
+            input_path: Path to the input file.
+
+        Raises:
+            CompileError: If compilation fails.
 
-    def compile_input_path(self, input_path, comp_obj, ext_vars, **kwargs):
-        """
-        Compile validated input_path in comp_obj
-        kwargs are passed into compile_file()
         """
-        target_name = ext_vars.target
+        target_name = self.target_name
         output_path = comp_obj.output_path
-        output_type = comp_obj.output_type
-        prune_output = comp_obj.prune
-        ext_vars_dict = ext_vars.model_dump(by_alias=True)
 
         logger.debug("Compiling %s", input_path)
         try:
-            if kwargs.get("compose_target_name", False):
-                _compile_path = os.path.join(self.compile_path, target_name.replace(".", "/"), output_path)
-            else:
-                _compile_path = os.path.join(self.compile_path, target_name, output_path)
+            target_compile_path = os.path.join(self.compile_path, target_name.replace(".", "/"), output_path)
+            os.makedirs(target_compile_path, exist_ok=True)
+
             self.compile_file(
+                comp_obj,
                 input_path,
-                _compile_path,
-                ext_vars_dict,
-                output=output_type,
-                target_name=target_name,
-                prune_output=prune_output,
-                **kwargs,
+                target_compile_path,
             )
 
         except KapitanError as e:
             raise CompileError("{}\nCompile error: failed to compile target: {}".format(e, target_name))
 
-    def make_compile_dirs(self, target_name, output_path, **kwargs):
-        """make compile dirs, skips if dirs exist"""
-        _compile_path = os.path.join(self.compile_path, target_name, output_path)
-        if kwargs.get("compose_target_name", False):
-            _compile_path = _compile_path.replace(".", "/")
+    @abc.abstractmethod
+    def compile_file(self, config: CompileInputTypeConfig, input_path: str, compile_path: str):
+        """Compile a single input file.
 
-        os.makedirs(_compile_path, exist_ok=True)
+        Args:
+            config: CompileInputTypeConfig object.
+            input_path: Path to the input file.
+            compile_path: Path to the output directory.
 
-    def compile_file(self, file_path, compile_path, ext_vars, **kwargs):
-        """implements compilation for file_path to compile_path with ext_vars"""
-        return NotImplementedError
+        Raises:
+            NotImplementedError: This is an abstract method.
 
-    def default_output_type(self):
-        "returns default output_type value"
+        """
         return NotImplementedError
 
 

kapitan/inputs/copy.py

@@ -10,41 +10,46 @@
 import shutil
 
 from kapitan.inputs.base import InputType
+from kapitan.inventory.model.input_types import KapitanInputTypeCopyConfig
 from kapitan.utils import copy_tree
 
 logger = logging.getLogger(__name__)
 
 
 class Copy(InputType):
-    def __init__(self, compile_path, search_paths, ref_controller, ignore_missing=False):
-        self.ignore_missing = ignore_missing
-        super().__init__("copy", compile_path, search_paths, ref_controller)
+    def compile_file(self, config: KapitanInputTypeCopyConfig, input_path, compile_path):
+        """Copy input_path to compile_path.
 
-    def compile_file(self, file_path, compile_path, ext_vars, **kwargs):
-        """
-        Write items in path as plain rendered files to compile_path.
-        path can be either a file or directory.
+        Args:
+            config (KapitanInputTypeCopyConfig): input configuration.
+            input_path (str): path to the file or directory to copy.
+            compile_path (str): path to the destination directory.
+
+        Raises:
+            OSError: if input_path does not exist and ignore_missing is False.
         """
 
-        # Whether to fail silently if the path does not exists.
-        ignore_missing = self.ignore_missing
+        ignore_missing = config.ignore_missing
+
         try:
-            logger.debug("Copying %s to %s.", file_path, compile_path)
-            if os.path.exists(file_path):
-                if os.path.isfile(file_path):
+            if os.path.exists(input_path):
+                logger.debug("Copying '%s' to '%s'.", input_path, compile_path)
+                if os.path.isfile(input_path):
                     if os.path.isfile(compile_path):
-                        shutil.copy2(file_path, compile_path)
+                        # overwrite existing file
+                        shutil.copy2(input_path, compile_path)
                     else:
+                        # create destination directory if it doesn't exist
                         os.makedirs(compile_path, exist_ok=True)
-                        shutil.copy2(file_path, os.path.join(compile_path, os.path.basename(file_path)))
+                        # copy file to destination directory
+                        shutil.copy2(input_path, os.path.join(compile_path, os.path.basename(input_path)))
                 else:
+                    # Resolve relative paths to avoid issues with copy_tree
                     compile_path = os.path.abspath(compile_path)  # Resolve relative paths
-                    copy_tree(file_path, compile_path)
-            elif ignore_missing == False:
-                raise OSError(f"Path {file_path} does not exist and `ignore_missing` is {ignore_missing}")
+                    copy_tree(input_path, compile_path)
+            elif not ignore_missing:
+                # Raise exception if input path does not exist and ignore_missing is False
+                raise OSError(f"Path {input_path} does not exist and `ignore_missing` is {ignore_missing}")
         except OSError as e:
+            # Log exception and re-raise
             logger.exception("Input dir not copied. Error: %s", e)
-
-    def default_output_type(self):
-        # no output_type options for copy
-        return None