Define Pydantic model to representing Kubernetes manifest

kpops/api/__init__.py

@@ -16,7 +16,7 @@
 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.components.common.kubernetes_model import KubernetesManifest
 from kpops.config import KpopsConfig
 from kpops.pipeline import (
     Pipeline,
@@ -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.components.common.kubernetes_model 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

@@ -1,8 +1,8 @@
 import logging
 from collections.abc import Iterable, Iterator
+from typing import Any
 
 from kpops.component_handlers.helm_wrapper.model import HelmDiffConfig, HelmTemplate
-from kpops.component_handlers.kubernetes.model 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[dict[str, Any], dict[str, Any]]]:
         """Compare 2 releases and generate a Change object for each difference.
 
         :param current_release: Iterable containing HelmTemplate objects for the current release
@@ -32,13 +32,16 @@ def calculate_changes(
             # get corresponding dry-run release
             new_resource = new_release_index.pop(current_resource.filepath, None)
             yield Change(
-                current_resource.manifest,
-                new_resource.manifest if new_resource else KubernetesManifest(),
+                current_resource.manifest.model_dump(),
+                new_resource.manifest.model_dump() if new_resource else {},
             )
 
         # collect added files
         for new_resource in new_release_index.values():
-            yield Change(KubernetesManifest(), new_resource.manifest)
+            yield Change(
+                {},
+                new_resource.manifest.model_dump(mode="json"),
+            )
 
     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,
+                change.new_value,
                 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.components.common.kubernetes_model 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.components.common.kubernetes_model import K8S_LABEL_MAX_LEN
 
 RELEASE_NAME_MAX_LEN = 53
 

kpops/components/base_components/helm_app.py

@@ -22,12 +22,14 @@
     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.components.common.kubernetes_model import (
+    K8S_LABEL_MAX_LEN,
+    KubernetesManifest,
+)
 from kpops.config import get_config
 from kpops.manifests.argo import ArgoSyncWave, enrich_annotations
 from kpops.utils.colorify import magentaify
@@ -144,7 +146,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,10 +13,10 @@
     FromTopic,
     InputTopicTypes,
 )
-from kpops.components.base_components.models.resource import Resource
 from kpops.components.base_components.models.to_section import (
     ToSection,
 )
+from kpops.components.common.kubernetes_model import KubernetesManifest
 from kpops.components.common.topic import (
     KafkaTopic,
     OutputTopicTypes,
@@ -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/common/kubernetes_model.py

@@ -1,12 +1,87 @@
 import enum
+from collections.abc import Iterator
+from typing import Any
 
-from pydantic import Field
+import yaml
+from pydantic import ConfigDict, Field
+from typing_extensions import override
 
 from kpops.utils.docstring import describe_attr
-from kpops.utils.pydantic import DescConfigModel
+from kpops.utils.pydantic import CamelCaseConfigModel, DescConfigModel
 
 # Matches plain integer or numbers with valid suffixes: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#meaning-of-memory
 MEMORY_PATTERN = r"^\d+([EPTGMk]|Ei|Pi|Ti|Gi|Mi|Ki)?$"
+K8S_LABEL_MAX_LEN = 63
+
+
+class ManagedFieldsEntry(CamelCaseConfigModel):
+    # Define this class based on its actual structure
+    fields: dict[str, Any] | None = None
+
+
+class OwnerReference(CamelCaseConfigModel):
+    # Define this class based on its actual structure
+    apiVersion: str
+    kind: str
+    name: str
+    uid: str
+    controller: bool | None = None
+    block_owner_deletion: bool | None = None
+
+
+class ObjectMeta(CamelCaseConfigModel):
+    """Metadata for all Kubernetes objects."""
+
+    annotations: dict[str, str] | None = None
+    creation_timestamp: str | None = Field(
+        None, description="Timestamp in RFC3339 format"
+    )
+    deletion_grace_period_seconds: int | None = None
+    deletion_timestamp: str | None = Field(
+        None, description="Timestamp in RFC3339 format"
+    )
+    finalizers: list[str] | None = None
+    generate_name: str | None = None
+    generation: int | None = None
+    labels: dict[str, str] | None = None
+    managed_fields: list[ManagedFieldsEntry] | None = None
+    name: str | None = None
+    namespace: str | None = None
+    owner_references: list[OwnerReference] | None = None
+    resource_version: str | None = None
+    self_link: str | None = Field(
+        None,
+        description="Deprecated field, not populated by Kubernetes in modern versions",
+    )
+    uid: str | None = None
+
+    model_config = ConfigDict(extra="allow")
+
+
+class KubernetesManifest(CamelCaseConfigModel):
+    api_version: str
+    kind: str
+    metadata: ObjectMeta
+
+    model_config = ConfigDict(extra="allow")
+
+    @classmethod
+    def from_yaml(
+        cls, /, content: str
+    ) -> Iterator["KubernetesManifest"]:  # TODO: typing.Self for Python 3.11+
+        manifests: Iterator[dict[str, Any]] = yaml.load_all(content, yaml.Loader)
+        for manifest in manifests:
+            yield cls(**manifest)
+
+    @override
+    def model_dump(self, **_: Any) -> dict[str, Any]:
+        return super().model_dump(
+            mode="json",
+            by_alias=True,
+            exclude_none=True,
+            exclude_defaults=True,
+            exclude_unset=True,
+        )
 
 
 class ServiceType(str, enum.Enum):

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -6,8 +6,8 @@
 
 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.kubernetes_model import KubernetesManifest
 from kpops.components.common.topic import (
     KafkaTopic,
     OutputTopicTypes,
@@ -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,8 +8,8 @@
 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.kubernetes_model import KubernetesManifest
 from kpops.components.common.topic import KafkaTopic
 from kpops.components.streams_bootstrap.base import (
     StreamsBootstrap,
@@ -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 ()