Merge pull request #1178 from kapicorp/refactor_inventory

Refactor inventory

compiled/minikube-es/script.sh

@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+
+set -ex
+
+compile_dir=$1
+
+echo "This is going into a file" > "${compile_dir}/${FILE_NAME}"

kapitan/inventory/inventory.py

@@ -7,119 +7,91 @@
 
 import logging
 import os
-import time
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from typing import overload, Union
-
+from pydantic import BaseModel, Field
 from kapitan.errors import KapitanError
-from reclass.values import item
-
+from typing import Annotated, Dict, Any, Optional
 logger = logging.getLogger(__name__)
 
 
-@dataclass
-class InventoryTarget:
-    name: str
-    path: str
-    parameters: dict = field(default_factory=dict)
-    classes: list = field(default_factory=list)
-    applications: list = field(default_factory=list)
-    exports: list = field(default_factory=dict)
+class InventoryTarget(BaseModel):
+    name: str = Field(exclude=True)
+    path: str = Field(exclude=True)
+    parameters: dict = dict()
+    classes: list = list()
+    applications: list = list()
+    exports: list = list()
 
 
 class Inventory(ABC):
-    _default_path: str = "inventory"
-
-    def __init__(self, inventory_path: str = _default_path, compose_target_name: bool = False):
+    def __init__(self, inventory_path: str = "inventory", compose_target_name: bool = False, ignore_class_notfound=False):
         self.inventory_path = inventory_path
-        self.targets_path = os.path.join(inventory_path, "targets")
-        self.classes_path = os.path.join(inventory_path, "classes")
-
-        # config
         self.compose_target_name = compose_target_name
-
-        self.targets = {}
-
+        self.targets_path = os.path.join(self.inventory_path, 'targets')
+        self.classes_path = os.path.join(self.inventory_path, 'classes')
+        self.initialised: bool = False
+        self.targets: dict[str, InventoryTarget] = {}
+
+        self.__initialise(ignore_class_notfound=ignore_class_notfound)
+
     @property
     def inventory(self) -> dict:
         """
         get all targets from inventory
-        targets will be rendered
         """
 
-        return {
-            target.name: {"parameters": target.parameters, "classes": target.classes}
-            for target in self.get_targets().values()
-        }
+        return self.targets
 
-    def search_targets(self) -> dict:
+    def __initialise(self, ignore_class_notfound) -> bool:
         """
-        look for targets at '<inventory_path>/targets/' and return targets without rendering parameters
+        look for targets at '<inventory_path>/targets/' and initialise them.
         """
-
-        for root, dirs, files in os.walk(self.targets_path):
-            for file in files:
-                # split file extension and check if yml/yaml
-                path = os.path.relpath(os.path.join(root, file), self.targets_path)
-
-                if self.compose_target_name:
-                    name, ext = os.path.splitext(path)
-                    name = name.replace(os.sep, ".")
-                else:
-                    name, ext = os.path.splitext(file)
+        if not self.initialised:
+            for root, dirs, files in os.walk(self.targets_path):
+                for file in files:
+                    # split file extension and check if yml/yaml
+                    path = os.path.relpath(os.path.join(root, file), self.targets_path)
+
+                    if self.compose_target_name:
+                        name, ext = os.path.splitext(path)
+                        name = name.replace(os.sep, ".")
+                    else:
+                        name, ext = os.path.splitext(file)
+
+                    if ext not in (".yml", ".yaml"):
+                        logger.debug(f"ignoring {file}: targets have to be .yml or .yaml files.")
+                        continue
+
+                    target = InventoryTarget(name=name, path=path)
+
+                    if self.targets.get(target.name):
+                        raise InventoryError(
+                            f"Conflicting targets {target.name}: {target.path} and {self.targets[target.name].path}. "
+                            f"Consider using '--compose-target-name'."
+                        )
 
-                if ext not in (".yml", ".yaml"):
-                    logger.debug(f"ignoring {file}: targets have to be .yml or .yaml files.")
-                    continue
-
-                target = InventoryTarget(name, path)
-
-
-
-                if self.targets.get(target.name):
-                    raise InventoryError(
-                        f"Conflicting targets {target.name}: {target.path} and {self.targets[target.name].path}. "
-                        f"Consider using '--compose-target-name'."
-                    )
-
-                self.targets[target.name] = target
-        return self.targets
+                    self.targets[target.name] = target
+
+            self.render_targets(self.targets, ignore_class_notfound=ignore_class_notfound)
+            self.initialised = True
+        return self.initialised
 
     def get_target(self, target_name: str, ignore_class_not_found: bool = False) -> InventoryTarget:
         """
         helper function to get rendered InventoryTarget object for single target
         """
-        return self.get_targets([target_name], ignore_class_not_found)[target_name]
+        return self.targets.get(target_name)
 
     def get_targets(self, target_names: list[str] = [], ignore_class_not_found: bool = False) -> dict:
         """
         helper function to get rendered InventoryTarget objects for multiple targets
         """
-        if not self.targets:
-            self.search_targets()
-
-        targets_to_render = []
-        targets = {}
 
-        if not target_names:
-            targets = self.targets
+        if target_names:
+            return {target_name: self.targets[target_name] for target_name in target_names if target_name in self.targets}
         else:
-            try:
-                targets = { target_name : self.targets[target_name] for target_name in target_names }
-            except KeyError as e:
-                if not ignore_class_not_found:
-                    raise InventoryError(f"targets not found: {set(target_names)-set(self.targets)}" )
-
-        for target in targets.values():
-            if not target.parameters:
-                targets_to_render.append(target)
-
-        if targets_to_render:
-            self.render_targets(targets_to_render, ignore_class_not_found)
-
-        return self.targets
-
+            return self.targets
+
     def get_parameters(self, target_names: str | list[str], ignore_class_not_found: bool = False) -> dict:
         """
         helper function to get rendered parameters for single target or multiple targets

kapitan/refs/secrets/vaultkv.py

@@ -138,6 +138,7 @@ def _encrypt(self, data, encode_base64=False):
                 response = client.secrets.kv.v2.read_secret_version(
                     path=self.path,
                     mount_point=self.mount,
+                    raise_on_deleted_version=False,
                 )
                 secrets = response["data"]["data"]
         except InvalidPath:
@@ -203,6 +204,7 @@ def _decrypt(self):
                 response = client.secrets.kv.v2.read_secret_version(
                     path=secret_path,
                     mount_point=mount,
+                    raise_on_deleted_version=False,
                 )
                 return_data = response["data"]["data"][secret_key]
             client.adapter.close()

kapitan/resources.py

@@ -276,7 +276,7 @@ def inventory(search_paths: list, target_name: str = None, inventory_path: str =
 
     if target_name:
         target = inv.get_target(target_name)
-        return dataclasses.asdict(target)
+        return target.model_dump()
 
     return inv.inventory
 
@@ -304,7 +304,7 @@ def generate_inventory(args):
         sys.exit(1)
 
 
-def get_inventory(inventory_path) -> Inventory:
+def get_inventory(inventory_path, ignore_class_notfound: bool = False) -> Inventory:
     """
     generic inventory function that makes inventory backend pluggable
     default backend is reclass
@@ -329,13 +329,11 @@ def get_inventory(inventory_path) -> Inventory:
     inventory_backend: Inventory = None
 
     logger.debug(f"Using {backend_id} as inventory backend")
-    inventory_backend = backend(inventory_path, compose_target_name)
+    inventory_backend = backend(inventory_path=inventory_path, compose_target_name=compose_target_name, ignore_class_notfound=ignore_class_notfound)
 
     cached.inv = inventory_backend
     # migrate inventory to selected inventory backend
     if hasattr(cached.args, "migrate") and cached.args.migrate:
         inventory_backend.migrate()
 
-    inventory_backend.search_targets()
-
-    return inventory_backend
+    return cached.inv

kapitan/targets.py

@@ -362,29 +362,35 @@ def save_inv_cache(compile_path, targets):
                 yaml.dump(cached.inv_cache, stream=f, default_flow_style=False)
 
 
-def load_target_inventory(inventory_path, targets, ignore_class_notfound=False):
+def load_target_inventory(inventory_path, requested_targets, ignore_class_notfound=False):
     """returns a list of target objects from the inventory"""
     target_objs = []
-    inv = get_inventory(inventory_path)
+    inv = get_inventory(inventory_path, ignore_class_notfound)
 
     # if '-t' is set on compile, only loop through selected targets
-    if targets:
-        targets_list = targets
+    if requested_targets:
+        targets = inv.get_targets(requested_targets)
     else:
-        targets_list = inv.targets.keys()
+        targets = inv.targets
 
-    for target_name in targets_list:
+    for target_name, target in targets.items():
         try:
-            target_obj = inv.get_parameters(target_name, ignore_class_notfound).get("kapitan")
+            if not target.parameters:
+                if ignore_class_notfound:
+                    continue
+                else:
+                    raise InventoryError(f"InventoryError: {target_name}: parameters is empty")
+
+            kapitan_target_configs = inv.get_parameters(target_name, ignore_class_notfound).get("kapitan")
             # check if parameters.kapitan is empty
-            if not target_obj:
+            if not kapitan_target_configs:
                 raise InventoryError(f"InventoryError: {target_name}: parameters.kapitan has no assignment")
-            target_obj["target_full_path"] = inv.targets[target_name].name.replace(".", "/")
+            kapitan_target_configs["target_full_path"] = inv.targets[target_name].name.replace(".", "/")
             require_compile = not ignore_class_notfound
-            valid_target_obj(target_obj, require_compile)
-            validate_matching_target_name(target_name, target_obj, inventory_path)
+            valid_target_obj(kapitan_target_configs, require_compile)
+            validate_matching_target_name(target_name, kapitan_target_configs, inventory_path)
             logger.debug(f"load_target_inventory: found valid kapitan target {target_name}")
-            target_objs.append(target_obj)
+            target_objs.append(kapitan_target_configs)
         except KeyError:
             logger.debug(f"load_target_inventory: target {target_name} has no kapitan compile obj")
 

tests/test_compose_node_name.py

@@ -32,8 +32,8 @@ def test_compose_target_name(self):
 
         # ensure normal rendering works
         compose_target_name = True
-        inv = self.inventory(temp_inventory_dir, compose_target_name)
-        found_targets = inv.search_targets()
+        inv = self.inventory(inventory_path=temp_inventory_dir, compose_target_name=compose_target_name)
+        found_targets = inv.targets
         self.assertEqual(sorted(example_target_names), sorted(list(found_targets.keys())))
         # ensure that actual rendering finds the same nodes as `search_targets()`
         for t in example_target_names:
@@ -51,14 +51,13 @@ def test_compose_target_name(self):
 
         # ensure inventory detects name collision
         compose_target_name = False
-        inv = self.inventory(temp_inventory_dir, compose_target_name)
         with self.assertRaises(InventoryError):
-            inv.search_targets()
+            inv = self.inventory(inventory_path=temp_inventory_dir, compose_target_name=compose_target_name)
 
         # ensure compose_target_name works as intended
         compose_target_name = True
-        inv = self.inventory(temp_inventory_dir, compose_target_name)
-        found_targets = inv.search_targets()
+        inv = self.inventory(inventory_path=temp_inventory_dir, compose_target_name=compose_target_name)
+        found_targets = inv.targets
 
         self.assertEqual(set(composed_target_names), set(found_targets.keys()))
         # ensure that actual rendering finds the same nodes as `search_targets()`