Define Pydantic model to representing Kubernetes manifest (#546)

Closes #548

kpops/api/__init__.py

@@ -16,8 +16,8 @@
 from kpops.component_handlers.schema_handler.schema_handler import SchemaHandler
 from kpops.component_handlers.topic.handler import TopicHandler
 from kpops.component_handlers.topic.proxy_wrapper import ProxyWrapper
-from kpops.components.base_components.models.resource import Resource
 from kpops.config import KpopsConfig
+from kpops.manifests.kubernetes import KubernetesManifest
 from kpops.pipeline import (
     Pipeline,
     PipelineGenerator,
@@ -79,7 +79,7 @@ def manifest_deploy(
     environment: str | None = None,
     verbose: bool = True,
     operation_mode: OperationMode = OperationMode.MANIFEST,
-) -> Iterator[Resource]:
+) -> Iterator[tuple[KubernetesManifest, ...]]:
     pipeline = generate(
         pipeline_path=pipeline_path,
         dotenv=dotenv,

kpops/cli/main.py

@@ -227,7 +227,7 @@ def deploy(
                 )
                 for resource in resources:
                     for rendered_manifest in resource:
-                        print_yaml(rendered_manifest)
+                        print_yaml(rendered_manifest.model_dump())
 
 
 @app.command(help="Destroy pipeline steps")

kpops/component_handlers/helm_wrapper/helm.py

@@ -20,13 +20,11 @@
     RepoAuthFlags,
     Version,
 )
-from kpops.component_handlers.kubernetes.model import KubernetesManifest
+from kpops.manifests.kubernetes import KubernetesManifest
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator
 
-    from kpops.components.base_components.models.resource import Resource
-
 
 log = logging.getLogger("Helm")
 
@@ -162,7 +160,7 @@ def template(
         namespace: str,
         values: dict[str, Any],
         flags: HelmTemplateFlags | None = None,
-    ) -> Resource:
+    ) -> tuple[KubernetesManifest, ...]:
         """From Helm: Render chart templates locally and display the output.
 
         Any values that would normally be looked up or retrieved in-cluster will
@@ -193,7 +191,7 @@ def template(
             command.extend(flags.to_command())
             output = self.__execute(command)
             manifests = KubernetesManifest.from_yaml(output)
-            return list(manifests)
+            return tuple(manifests)
 
     def get_manifest(self, release_name: str, namespace: str) -> Iterable[HelmTemplate]:
         command = [

kpops/component_handlers/helm_wrapper/helm_diff.py

@@ -2,7 +2,7 @@
 from collections.abc import Iterable, Iterator
 
 from kpops.component_handlers.helm_wrapper.model import HelmDiffConfig, HelmTemplate
-from kpops.component_handlers.kubernetes.model import KubernetesManifest
+from kpops.manifests.kubernetes import KubernetesManifest
 from kpops.utils.dict_differ import Change, render_diff
 
 log = logging.getLogger("HelmDiff")
@@ -16,7 +16,7 @@ def __init__(self, config: HelmDiffConfig) -> None:
     def calculate_changes(
         current_release: Iterable[HelmTemplate],
         new_release: Iterable[HelmTemplate],
-    ) -> Iterator[Change[KubernetesManifest, KubernetesManifest]]:
+    ) -> Iterator[Change[KubernetesManifest | None, KubernetesManifest | None]]:
         """Compare 2 releases and generate a Change object for each difference.
 
         :param current_release: Iterable containing HelmTemplate objects for the current release
@@ -33,12 +33,15 @@ def calculate_changes(
             new_resource = new_release_index.pop(current_resource.filepath, None)
             yield Change(
                 current_resource.manifest,
-                new_resource.manifest if new_resource else KubernetesManifest(),
+                new_resource.manifest if new_resource else None,
             )
 
         # collect added files
         for new_resource in new_release_index.values():
-            yield Change(KubernetesManifest(), new_resource.manifest)
+            yield Change(
+                None,
+                new_resource.manifest,
+            )
 
     def log_helm_diff(
         self,
@@ -48,8 +51,8 @@ def log_helm_diff(
     ) -> None:
         for change in self.calculate_changes(current_release, new_release):
             if diff := render_diff(
-                change.old_value.data,
-                change.new_value.data,
+                change.old_value.model_dump() if change.old_value else {},
+                change.new_value.model_dump() if change.new_value else {},
                 ignore=self.config.ignore,
             ):
                 logger.info("\n" + diff)

kpops/component_handlers/helm_wrapper/model.py

@@ -6,7 +6,7 @@
 from typing_extensions import override
 
 from kpops.component_handlers.helm_wrapper.exception import ParseError
-from kpops.component_handlers.kubernetes.model import KubernetesManifest
+from kpops.manifests.kubernetes import KubernetesManifest
 from kpops.utils.docstring import describe_attr
 from kpops.utils.pydantic import DescConfigModel
 

kpops/component_handlers/helm_wrapper/utils.py

@@ -1,5 +1,5 @@
-from kpops.component_handlers.kubernetes.model import K8S_LABEL_MAX_LEN
 from kpops.component_handlers.kubernetes.utils import trim
+from kpops.manifests.kubernetes import K8S_LABEL_MAX_LEN
 
 RELEASE_NAME_MAX_LEN = 53
 

kpops/components/base_components/helm_app.py

@@ -22,14 +22,13 @@
     create_helm_name_override,
     create_helm_release_name,
 )
-from kpops.component_handlers.kubernetes.model import K8S_LABEL_MAX_LEN
 from kpops.components.base_components.kubernetes_app import (
     KubernetesApp,
     KubernetesAppValues,
 )
-from kpops.components.base_components.models.resource import Resource
 from kpops.config import get_config
 from kpops.manifests.argo import ArgoSyncWave, enrich_annotations
+from kpops.manifests.kubernetes import K8S_LABEL_MAX_LEN, KubernetesManifest
 from kpops.utils.colorify import magentaify
 from kpops.utils.docstring import describe_attr
 from kpops.utils.pydantic import exclude_by_name
@@ -144,7 +143,7 @@ def template_flags(self) -> HelmTemplateFlags:
         )
 
     @override
-    def manifest_deploy(self) -> Resource:
+    def manifest_deploy(self) -> tuple[KubernetesManifest, ...]:
         values = self.to_helm_values()
         if get_config().operation_mode is OperationMode.ARGO:
             sync_wave = ArgoSyncWave(sync_wave=1)

kpops/components/base_components/models/resource.py

@@ -1,5 +0,0 @@
-from collections.abc import Mapping
-from typing import Any, TypeAlias
-
-# representation of final resource for component, e.g. a list of Kubernetes manifests
-Resource: TypeAlias = list[Mapping[str, Any]]

kpops/components/base_components/pipeline_component.py

@@ -13,7 +13,6 @@
     FromTopic,
     InputTopicTypes,
 )
-from kpops.components.base_components.models.resource import Resource
 from kpops.components.base_components.models.to_section import (
     ToSection,
 )
@@ -22,6 +21,7 @@
     OutputTopicTypes,
     TopicConfig,
 )
+from kpops.manifests.kubernetes import KubernetesManifest
 from kpops.utils.docstring import describe_attr
 
 
@@ -229,21 +229,21 @@ def inflate(self) -> list[PipelineComponent]:
         """
         return [self]
 
-    def manifest_deploy(self) -> Resource:
+    def manifest_deploy(self) -> tuple[KubernetesManifest, ...]:
         """Render Kubernetes manifests for deploy."""
-        return []
+        return ()
 
-    def manifest_destroy(self) -> Resource:
+    def manifest_destroy(self) -> tuple[KubernetesManifest, ...]:
         """Render Kubernetes manifests resources for destroy."""
-        return []
+        return ()
 
-    def manifest_reset(self) -> Resource:
+    def manifest_reset(self) -> tuple[KubernetesManifest, ...]:
         """Render Kubernetes manifests resources for reset."""
-        return []
+        return ()
 
-    def manifest_clean(self) -> Resource:
+    def manifest_clean(self) -> tuple[KubernetesManifest, ...]:
         """Render Kubernetes manifests resources for clean."""
-        return []
+        return ()
 
     async def deploy(self, dry_run: bool) -> None:
         """Deploy component, e.g. to Kubernetes cluster.

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -6,7 +6,6 @@
 
 from kpops.api import OperationMode
 from kpops.components.base_components.kafka_app import KafkaAppCleaner
-from kpops.components.base_components.models.resource import Resource
 from kpops.components.common.app_type import AppType
 from kpops.components.common.topic import (
     KafkaTopic,
@@ -20,6 +19,7 @@
 from kpops.config import get_config
 from kpops.const.file_type import DEFAULTS_YAML, PIPELINE_YAML
 from kpops.manifests.argo import ArgoHook, enrich_annotations
+from kpops.manifests.kubernetes import KubernetesManifest
 from kpops.utils.docstring import describe_attr
 
 log = logging.getLogger("ProducerApp")
@@ -36,11 +36,12 @@ def helm_chart(self) -> str:
         )
 
     @override
-    def manifest_deploy(self) -> Resource:
+    def manifest_deploy(self) -> tuple[KubernetesManifest, ...]:
         values = self.to_helm_values()
         if get_config().operation_mode is OperationMode.ARGO:
             post_delete = ArgoHook.POST_DELETE
             values = enrich_annotations(values, post_delete.key, post_delete.value)
+
         return self.helm.template(
             self.helm_release_name,
             self.helm_chart,
@@ -144,11 +145,12 @@ async def clean(self, dry_run: bool) -> None:
         await super().clean(dry_run)
         await self._cleaner.clean(dry_run)
 
-    def manifest_deploy(self) -> Resource:
+    @override
+    def manifest_deploy(self) -> tuple[KubernetesManifest, ...]:
         manifests = super().manifest_deploy()
         operation_mode = get_config().operation_mode
 
         if operation_mode is OperationMode.ARGO:
-            manifests.extend(self._cleaner.manifest_deploy())
+            manifests = manifests + self._cleaner.manifest_deploy()
 
         return manifests

kpops/components/streams_bootstrap/streams/streams_app.py

@@ -8,7 +8,6 @@
 from kpops.component_handlers.kubernetes.pvc_handler import PVCHandler
 from kpops.components.base_components.helm_app import HelmApp
 from kpops.components.base_components.kafka_app import KafkaAppCleaner
-from kpops.components.base_components.models.resource import Resource
 from kpops.components.common.app_type import AppType
 from kpops.components.common.topic import KafkaTopic
 from kpops.components.streams_bootstrap.base import (
@@ -20,6 +19,7 @@
 from kpops.config import get_config
 from kpops.const.file_type import DEFAULTS_YAML, PIPELINE_YAML
 from kpops.manifests.argo import ArgoHook, enrich_annotations
+from kpops.manifests.kubernetes import KubernetesManifest
 from kpops.utils.docstring import describe_attr
 
 log = logging.getLogger("StreamsApp")
@@ -49,7 +49,7 @@ async def clean(self, dry_run: bool) -> None:
             await self.clean_pvcs(dry_run)
 
     @override
-    def manifest_deploy(self) -> Resource:
+    def manifest_deploy(self) -> tuple[KubernetesManifest, ...]:
         values = self.to_helm_values()
         if get_config().operation_mode is OperationMode.ARGO:
             post_delete = ArgoHook.POST_DELETE
@@ -174,13 +174,13 @@ async def clean(self, dry_run: bool) -> None:
         await self._cleaner.clean(dry_run)
 
     @override
-    def manifest_deploy(self) -> Resource:
+    def manifest_deploy(self) -> tuple[KubernetesManifest, ...]:
         manifests = super().manifest_deploy()
         if get_config().operation_mode is OperationMode.ARGO:
-            manifests.extend(self._cleaner.manifest_deploy())
+            manifests = manifests + self._cleaner.manifest_deploy()
 
         return manifests
 
     @override
-    def manifest_clean(self) -> Resource:
-        return []
+    def manifest_clean(self) -> tuple[KubernetesManifest, ...]:
+        return ()