Adds validation for Inventory with Pydantic (#1234)

Fixes #

## Proposed Changes

* Adds inventory validation using Pydantic

## Docs and Tests

* [X] Tests added

examples/kubernetes/compiled/jsonnet-env/jsonnet-env/env.yml

@@ -19,22 +19,32 @@ parameters:
   c: ccccc
   kapitan:
     compile:
-      - input_params: {}
+      - continue_on_compile_error: false
+        ignore_missing: false
+        input_params: {}
         input_paths:
           - components/jsonnet-env/env.jsonnet
         input_type: jsonnet
+        name: null
         output_path: jsonnet-env
         output_type: yml
+        prune: false
+    dependencies: []
+    labels: {}
     secrets:
       awskms:
         key: alias/nameOfKey
+      azkms: null
       gkms:
         key: projects/<project>/locations/<location>/keyRings/<keyRing>/cryptoKeys/<key>
       gpg:
         recipients:
           - fingerprint: D9234C61F58BEB3ED8552A57E28DC07A3CBFAE7C
             name: [email protected]
+      vaultkv: null
+      vaulttransit: null
     target_full_path: jsonnet-env
+    validate: []
     vars:
       managed_by: kapitan
       namespace: jsonnet-env

kapitan/dependency_manager/base.py

@@ -15,6 +15,7 @@
 
 from kapitan.errors import GitFetchingError, GitSubdirNotFoundError, HelmFetchingError
 from kapitan.helm_cli import helm_cli
+from kapitan.inventory.model.dependencies import KapitanDependencyTypes
 from kapitan.utils import (
     copy_tree,
     make_request,
@@ -46,16 +47,16 @@ def fetch_dependencies(output_path, target_objs, save_dir, force, pool):
     deps_output_paths = defaultdict(set)
     for target_obj in target_objs:
         try:
-            dependencies = target_obj["dependencies"]
+            dependencies = target_obj.dependencies
             for item in dependencies:
-                dependency_type = item["type"]
-                source_uri = item["source"]
+                dependency_type = item.type
+                source_uri = item.source
 
                 # The target key "output_path" is relative to the compile output path set by the user
                 # point to the full output path
-                full_output_path = normalise_join_path(output_path, item["output_path"])
-                logger.debug("Updated output_path from %s to %s", item["output_path"], output_path)
-                item["output_path"] = full_output_path
+                full_output_path = normalise_join_path(output_path, item.output_path)
+                logger.debug("Updated output_path from %s to %s", item.output_path, output_path)
+                item.output_path = full_output_path
 
                 if full_output_path in deps_output_paths[source_uri]:
                     # if the output_path is duplicated for the same source_uri
@@ -64,15 +65,13 @@ def fetch_dependencies(output_path, target_objs, save_dir, force, pool):
                 else:
                     deps_output_paths[source_uri].add(full_output_path)
 
-                if dependency_type == "git":
+                if dependency_type == KapitanDependencyTypes.GIT:
                     git_deps[source_uri].append(item)
-                elif dependency_type in ("http", "https"):
+                elif dependency_type in (KapitanDependencyTypes.HTTP, KapitanDependencyTypes.HTTPS):
                     http_deps[source_uri].append(item)
-                elif dependency_type == "helm":
-                    version = item.get("version", "")
-                    helm_deps[
-                        HelmSource(source_uri, item["chart_name"], version, item.get("helm_path"))
-                    ].append(item)
+                elif dependency_type == KapitanDependencyTypes.HELM:
+                    version = item.version
+                    helm_deps[HelmSource(source_uri, item.chart_name, version, item.helm_path)].append(item)
                 else:
                     logger.warning("%s is not a valid source type", dependency_type)
 
@@ -106,21 +105,17 @@ def fetch_git_dependency(dep_mapping, save_dir, force, item_type="Dependency"):
         logger.debug("Using cached %s %s", item_type, cached_repo_path)
     for dep in deps:
         repo = Repo(cached_repo_path)
-        output_path = dep["output_path"]
+        output_path = dep.output_path
         copy_src_path = cached_repo_path
-        if "ref" in dep:
-            ref = dep["ref"]
-            repo.git.checkout(ref)
-        else:
-            repo.git.checkout("master")  # default ref
+        repo.git.checkout(dep.ref)
 
         # initialising submodules
-        if "submodules" not in dep or dep["submodules"]:
+        if dep.submodules:
             for submodule in repo.submodules:
                 submodule.update(init=True)
 
-        if "subdir" in dep:
-            sub_dir = dep["subdir"]
+        if dep.subdir:
+            sub_dir = dep.subdir
             full_subdir = os.path.join(cached_repo_path, sub_dir)
             if os.path.isdir(full_subdir):
                 copy_src_path = full_subdir
@@ -169,8 +164,8 @@ def fetch_http_dependency(dep_mapping, save_dir, force, item_type="Dependency"):
         logger.debug("Using cached %s %s", item_type, cached_source_path)
         content_type = MimeTypes().guess_type(cached_source_path)[0]
     for dep in deps:
-        output_path = dep["output_path"]
-        if dep.get("unpack", False):
+        output_path = dep.output_path
+        if dep.unpack:
             # ensure that the directory we are extracting to exists
             os.makedirs(output_path, exist_ok=True)
             if force:
@@ -236,7 +231,7 @@ def fetch_helm_chart(dep_mapping, save_dir, force):
     )
 
     for dep in deps:
-        output_path = dep["output_path"]
+        output_path = dep.output_path
 
         if not os.path.exists(output_path) or force:
             if not exists_in_cache(cached_repo_path):

kapitan/errors.py

@@ -9,56 +9,38 @@
 class KapitanError(Exception):
     """generic kapitan error"""
 
-    pass
-
 
 class CompileError(KapitanError):
     """compile error"""
 
-    pass
-
 
 class InventoryError(KapitanError):
     """inventory error"""
 
-    pass
-
 
 class SecretError(KapitanError):
     """secrets error"""
 
-    pass
-
 
 class RefError(KapitanError):
     """ref error"""
 
-    pass
-
 
 class RefBackendError(KapitanError):
     """ref backend error"""
 
-    pass
-
 
 class RefFromFuncError(KapitanError):
     """ref from func error"""
 
-    pass
-
 
 class RefHashMismatchError(KapitanError):
     """ref has mismatch error"""
 
-    pass
-
 
 class HelmBindingUnavailableError(KapitanError):
     """helm input is used when the binding is not available"""
 
-    pass
-
 
 class HelmFetchingError(KapitanError):
     pass
@@ -71,22 +53,14 @@ class HelmTemplateError(KapitanError):
 class GitSubdirNotFoundError(KapitanError):
     """git dependency subdir not found error"""
 
-    pass
-
 
 class GitFetchingError(KapitanError):
     """repo not found and/or permission error"""
 
-    pass
-
 
 class RequestUnsuccessfulError(KapitanError):
     """request error"""
 
-    pass
-
 
 class KubernetesManifestValidationError(KapitanError):
     """kubernetes manifest schema validation error"""
-
-    pass

kapitan/inputs/base.py

@@ -36,17 +36,17 @@ def compile_obj(self, comp_obj, ext_vars, **kwargs):
         and run compile_input_path() for each resolved input_path.
         kwargs are passed into compile_input_path()
         """
-        input_type = comp_obj["input_type"]
+        input_type = comp_obj.input_type
         assert input_type == self.type_name
 
         # expand any globbed paths, taking into account provided search paths
         input_paths = []
-        for input_path in comp_obj["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))
             # remove duplicate inputs
             inputs = set(inputs)
-            ignore_missing = comp_obj.get("ignore_missing", False)
+            ignore_missing = comp_obj.ignore_missing
             if len(inputs) == 0 and not ignore_missing:
                 raise CompileError(
                     "Compile error: {} for target: {} not found in "
@@ -62,10 +62,11 @@ 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"]
-        output_path = comp_obj["output_path"]
-        output_type = comp_obj.get("output_type", self.default_output_type())
-        prune_output = comp_obj.get("prune", kwargs.get("prune", False))
+        target_name = ext_vars.target
+        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:
@@ -76,7 +77,7 @@ def compile_input_path(self, input_path, comp_obj, ext_vars, **kwargs):
             self.compile_file(
                 input_path,
                 _compile_path,
-                ext_vars,
+                ext_vars_dict,
                 output=output_type,
                 target_name=target_name,
                 prune_output=prune_output,

kapitan/inputs/external.py

@@ -8,7 +8,6 @@
 import logging
 import os
 import re
-import shutil
 import subprocess
 
 from kapitan.inputs.base import InputType

kapitan/inputs/helm.py

@@ -15,6 +15,7 @@
 from kapitan.helm_cli import helm_cli
 from kapitan.inputs.base import CompiledFile, InputType
 from kapitan.inputs.kadet import BaseModel, BaseObj, Dict
+from kapitan.inventory.model.input_types import KapitanInputTypeHelmConfig
 
 logger = logging.getLogger(__name__)
 
@@ -30,22 +31,19 @@
 
 
 class Helm(InputType):
-    def __init__(self, compile_path, search_paths, ref_controller, args):
+    def __init__(self, compile_path, search_paths, ref_controller, args: KapitanInputTypeHelmConfig):
         super().__init__("helm", compile_path, search_paths, ref_controller)
 
-        self.helm_values_files = args.get("helm_values_files")
-        self.helm_params = args.get("helm_params") or {}
-        self.helm_path = args.get("helm_path")
+        self.helm_values_files = args.helm_values_files
+        self.helm_params = args.helm_params
+        self.helm_path = args.helm_path
         self.file_path = None
 
         self.helm_values_file = None
-        if "helm_values" in args:
-            self.helm_values_file = write_helm_values_file(args["helm_values"])
+        if args.helm_values:
+            self.helm_values_file = write_helm_values_file(args.helm_values)
 
-        self.kube_version = None
-        if "kube_version" in args:
-            logger.warning("passing kube_version is deprecated. Use api_versions helm flag instead.")
-            self.kube_version = args["kube_version"]
+        self.kube_version = args.kube_version
 
     def compile_file(self, file_path, compile_path, ext_vars, **kwargs):
         """

kapitan/inputs/jinja2.py

@@ -19,11 +19,11 @@
 class Jinja2(InputType):
     def __init__(self, compile_path, search_paths, ref_controller, args):
         super().__init__("jinja2", compile_path, search_paths, ref_controller)
-        self.strip_postfix = args.get("suffix_remove", False)
-        self.stripped_postfix = args.get("suffix_stripped", ".j2")
-        self.input_params = args.get("input_params", {})
+        self.strip_postfix = args.suffix_remove
+        self.stripped_postfix = args.suffix_stripped
+        self.input_params = args.input_params
 
-    def compile_file(self, file_path, compile_path, ext_vars, **kwargs):
+    def compile_file(self, file_path, compile_path, ext_vars={}, **kwargs):
         """
         Write items in path as jinja2 rendered files to compile_path.
         path can be either a file or directory.
@@ -43,8 +43,8 @@ def compile_file(self, file_path, compile_path, ext_vars, **kwargs):
         # set ext_vars and inventory for jinja2 context
         context = ext_vars.copy()
 
-        context["inventory_global"] = cached.inv.inventory
-        context["inventory"] = cached.inv.inventory[target_name]
+        context["inventory_global"] = cached.global_inv
+        context["inventory"] = cached.global_inv[target_name]
         context["input_params"] = input_params
 
         jinja2_filters = kwargs.get("jinja2_filters")

kapitan/inputs/jsonnet.py

@@ -10,7 +10,6 @@
 import os
 import sys
 
-from kapitan import cached
 from kapitan.errors import CompileError
 from kapitan.inputs.base import CompiledFile, InputType
 from kapitan.resources import resource_callbacks, search_imports
@@ -81,6 +80,7 @@ def _search_imports(cwd, imp):
             return search_imports(cwd, imp, self.search_paths)
 
         json_output = None
+
         if self.use_go:
             json_output = go_jsonnet_file(
                 file_path,

kapitan/inputs/kadet.py

@@ -10,7 +10,6 @@
 import logging
 import os
 import sys
-import time
 from functools import lru_cache
 from importlib.util import module_from_spec, spec_from_file_location
 

kapitan/inventory/__init__.py

@@ -1,11 +1,8 @@
-from .inventory import Inventory
+from typing import Type
 
-try:
-    from enum import StrEnum
-except ImportError:
-    from strenum import StrEnum
+from kapitan.utils import StrEnum
 
-from typing import Type
+from .inventory import Inventory
 
 
 class InventoryBackends(StrEnum):