Handle rendering resources with multiple manifests

kpops/cli/main.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Iterator, Mapping
+from collections.abc import Iterator
 from enum import Enum
 from pathlib import Path
 from typing import TYPE_CHECKING, Optional
@@ -19,6 +19,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.pipeline_component import Resource
 from kpops.config import ENV_PREFIX, KpopsConfig
 from kpops.pipeline_generator.pipeline import Pipeline
 from kpops.utils.gen_schema import SchemaScope, gen_config_schema, gen_pipeline_schema
@@ -271,7 +272,7 @@ def render(
     filter_type: FilterType = FILTER_TYPE,
     output: bool = OUTPUT_OPTION,
     verbose: bool = VERBOSE_OPTION,
-) -> list[Mapping]:
+) -> list[Resource]:
     pipeline = generate(
         pipeline_path=pipeline_path,
         components_module=components_module,
@@ -282,13 +283,14 @@ def render(
         verbose=verbose,
     )
     steps_to_apply = get_steps_to_apply(pipeline, steps, filter_type)
-    manifests: list[Mapping] = []
+    resources: list[Resource] = []
     for component in steps_to_apply:
-        manifest = component.render()
-        manifests.append(manifest)
+        resource = component.render()
+        resources.append(resource)
         if output:
-            print_yaml(manifest)
-    return manifests
+            for manifest in resource:
+                print_yaml(manifest)
+    return resources
 
 
 @app.command(

kpops/component_handlers/helm_wrapper/helm.py

@@ -20,6 +20,7 @@
     Version,
 )
 from kpops.component_handlers.kubernetes.model import KubernetesManifest
+from kpops.components.base_components.pipeline_component import Resource
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator
@@ -134,7 +135,7 @@ def template(
         namespace: str,
         values: dict,
         flags: HelmTemplateFlags | None = None,
-    ) -> KubernetesManifest:
+    ) -> Resource:
         """From Helm: Render chart templates locally and display the output.
 
         Any values that would normally be looked up or retrieved in-cluster will
@@ -146,7 +147,7 @@ def template(
         :param namespace: The Kubernetes namespace the command should execute in
         :param values: `values.yaml` to be used
         :param flags: the flags to be set for `helm template`, defaults to HelmTemplateFlags()
-        :return: the rendered manifest
+        :return: the rendered resource (list of Kubernetes manifests)
         """
         if flags is None:
             flags = HelmTemplateFlags()
@@ -164,7 +165,8 @@ def template(
             ]
             command.extend(flags.to_command())
             output = self.__execute(command)
-            return KubernetesManifest.from_yaml(output)
+            manifests = KubernetesManifest.from_yaml(output)
+            return list(manifests)
 
     def get_manifest(self, release_name: str, namespace: str) -> Iterable[HelmTemplate]:
         command = [
@@ -201,7 +203,10 @@ def load_manifest(yaml_contents: str) -> Iterator[HelmTemplate]:
             if line.startswith("---"):
                 is_beginning = True
                 if template_name and current_yaml_doc:
-                    manifest = KubernetesManifest.from_yaml("\n".join(current_yaml_doc))
+                    manifests = KubernetesManifest.from_yaml(
+                        "\n".join(current_yaml_doc)
+                    )
+                    manifest = next(manifests)  # only 1 manifest
                     yield HelmTemplate(Path(template_name), manifest)
                     template_name = None
                     current_yaml_doc.clear()

kpops/component_handlers/kubernetes/model.py

@@ -1,5 +1,6 @@
 import json
 from collections import UserDict
+from collections.abc import Iterator
 from typing import TypeAlias
 
 import yaml
@@ -16,9 +17,10 @@ class KubernetesManifest(UserDict[str, Json]):
     """Representation of a Kubernetes API object as YAML/JSON mapping."""
 
     @classmethod
-    def from_yaml(cls, /, content: str) -> Self:
-        manifest: dict = yaml.load(content, yaml.Loader)
-        return cls(manifest)
+    def from_yaml(cls, /, content: str) -> Iterator[Self]:
+        manifests: Iterator[dict[str, Json]] = yaml.load_all(content, yaml.Loader)
+        for manifest in manifests:
+            yield cls(manifest)
 
     @classmethod
     def from_json(cls, /, content: str) -> Self:

kpops/components/base_components/helm_app.py

@@ -16,8 +16,8 @@
     HelmTemplateFlags,
     HelmUpgradeInstallFlags,
 )
-from kpops.component_handlers.kubernetes.model import KubernetesManifest
 from kpops.components.base_components.kubernetes_app import KubernetesApp
+from kpops.components.base_components.pipeline_component import Resource
 from kpops.utils.colorify import magentaify
 from kpops.utils.docstring import describe_attr
 
@@ -96,7 +96,7 @@ def template_flags(self) -> HelmTemplateFlags:
         )
 
     @override
-    def render(self) -> KubernetesManifest:
+    def render(self) -> Resource:
         return self.helm.template(
             self.helm_release_name,
             self.helm_chart,

kpops/components/base_components/kafka_connector.py

@@ -24,10 +24,12 @@
     KafkaConnectResetterConfig,
     KafkaConnectResetterValues,
 )
-from kpops.component_handlers.kubernetes.model import KubernetesManifest
 from kpops.components.base_components.base_defaults_component import deduplicate
 from kpops.components.base_components.models.from_section import FromTopic
-from kpops.components.base_components.pipeline_component import PipelineComponent
+from kpops.components.base_components.pipeline_component import (
+    PipelineComponent,
+    Resource,
+)
 from kpops.utils.colorify import magentaify
 from kpops.utils.docstring import describe_attr
 
@@ -285,7 +287,7 @@ def apply_from_inputs(self, name: str, topic: FromTopic) -> NoReturn:
         raise NotImplementedError(msg)
 
     @override
-    def render(self) -> KubernetesManifest:
+    def render(self) -> Resource:
         values = self._get_kafka_connect_resetter_values(
             offset_topic=self.offset_topic,
         )
@@ -331,7 +333,7 @@ def add_input_topics(self, topics: list[str]) -> None:
         setattr(self.app, "topics", ",".join(topics))
 
     @override
-    def render(self) -> KubernetesManifest:
+    def render(self) -> Resource:
         values = self._get_kafka_connect_resetter_values()
         return self.helm.template(
             self._resetter_release_name,

kpops/components/base_components/pipeline_component.py

@@ -2,6 +2,7 @@
 
 from abc import ABC
 from collections.abc import Mapping
+from typing import TypeAlias
 
 from pydantic import Extra, Field
 
@@ -21,6 +22,8 @@
 from kpops.utils.docstring import describe_attr
 from kpops.utils.pydantic import DescConfig
 
+Resource: TypeAlias = list[Mapping]  # representation of final resource  # TODO: move?
+
 
 class PipelineComponent(BaseDefaultsComponent, ABC):
     """Base class for all components.
@@ -187,9 +190,9 @@ def inflate(self) -> list[PipelineComponent]:
         """
         return [self]
 
-    def render(self) -> Mapping:
+    def render(self) -> Resource:
         """Render final component resources, e.g. Kubernetes manifest."""
-        return {}
+        return []
 
     def deploy(self, dry_run: bool) -> None:
         """Deploy component, e.g. to the Kubernetes cluster.

tests/component_handlers/kubernetes/model.py

@@ -21,15 +21,18 @@ class TestKubernetesManifest:
                             foo: bar
                     """
                 ),
-                KubernetesManifest(
-                    {
-                        "apiVersion": "v1",
-                        "kind": "ServiceAccount",
-                        "metadata": {"labels": {"foo": "bar"}},
-                    }
-                ),
+                [
+                    KubernetesManifest(
+                        {
+                            "apiVersion": "v1",
+                            "kind": "ServiceAccount",
+                            "metadata": {"labels": {"foo": "bar"}},
+                        }
+                    )
+                ],
             )
         ],
     )
     def test_from_yaml(self, helm_template: str, expected_manifest: KubernetesManifest):
-        assert KubernetesManifest.from_yaml(helm_template) == expected_manifest
+        manifests = KubernetesManifest.from_yaml(helm_template)
+        assert list(manifests) == expected_manifest

tests/pipeline/test_pipeline.py

@@ -23,6 +23,7 @@ def test_python_api_generate(self):
             "tests.pipeline.test_components",
             pipeline_base_dir=PIPELINE_BASE_DIR_PATH,
             defaults=RESOURCE_PATH,
+            output=False,
         )
         assert len(pipeline) == 3