Support networkx to represent components as a graph

kpops/components/base_components/kafka_connector.py

@@ -351,6 +351,11 @@ def __run_kafka_connect_resetter(self, dry_run: bool) -> None:
 class KafkaSinkConnector(KafkaConnector):
     """Kafka sink connector model"""
 
+    @override
+    def get_input_topics(self) -> list[str]:
+        topics = getattr(self.app, "topics", None)
+        return topics.split(",") if topics is not None else []
+
     @override
     def add_input_topics(self, topics: list[str]) -> None:
         existing_topics: str | None = getattr(self.app, "topics", None)

kpops/components/base_components/pipeline_component.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC
+from collections.abc import Iterator
 
 from pydantic import Extra, Field
 
@@ -58,6 +59,21 @@ def __init__(self, **kwargs) -> None:
         self.set_input_topics()
         self.set_output_topics()
 
+    def get_input_topics(self) -> list[str]:
+        """Get all the input topics from config."""
+        return []
+
+    def get_extra_input_topics(self) -> dict[str, list[str]]:
+        """Get extra input topics list from config."""
+        return {}
+
+    def get_output_topic(self) -> str | None:
+        """Get output topic from config."""
+
+    def get_extra_output_topics(self) -> dict[str, str]:
+        """Get extra output topics list from config."""
+        return {}
+
     @property
     def full_name(self) -> str:
         return self.prefix + self.name
@@ -116,6 +132,18 @@ def set_input_topics(self) -> None:
             for name, topic in self.from_.topics.items():
                 self.apply_from_inputs(name, topic)
 
+    @property
+    def inputs(self) -> Iterator[str]:
+        yield from self.get_input_topics()
+        for role_topics in self.get_extra_input_topics().values():
+            yield from role_topics
+
+    @property
+    def outputs(self) -> Iterator[str]:
+        if output_topic := self.get_output_topic():
+            yield output_topic
+        yield from self.get_extra_output_topics().values()
+
     def apply_from_inputs(self, name: str, topic: FromTopic) -> None:
         """Add a `from` section input to the component config
 

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -44,6 +44,14 @@ def apply_to_outputs(self, name: str, topic: TopicConfig) -> None:
             case _:
                 super().apply_to_outputs(name, topic)
 
+    @override
+    def get_output_topic(self) -> str | None:
+        return self.app.streams.output_topic
+
+    @override
+    def get_extra_output_topics(self) -> dict[str, str]:
+        return self.app.streams.extra_output_topics
+
     @override
     def set_output_topic(self, topic_name: str) -> None:
         self.app.streams.output_topic = topic_name

kpops/components/streams_bootstrap/streams/streams_app.py

@@ -20,6 +20,22 @@ class StreamsApp(KafkaApp):
         description=describe_attr("app", __doc__),
     )
 
+    @override
+    def get_input_topics(self) -> list[str]:
+        return self.app.streams.input_topics
+
+    @override
+    def get_extra_input_topics(self) -> dict[str, list[str]]:
+        return self.app.streams.extra_input_topics
+
+    @override
+    def get_output_topic(self) -> str | None:
+        return self.app.streams.output_topic
+
+    @override
+    def get_extra_output_topics(self) -> dict[str, str] | None:
+        return self.app.streams.extra_output_topics
+
     @override
     def add_input_topics(self, topics: list[str]) -> None:
         self.app.streams.add_input_topics(topics)

kpops/pipeline_generator/pipeline.py

@@ -7,8 +7,9 @@
 from contextlib import suppress
 from pathlib import Path
 
+import networkx as nx
 import yaml
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from rich.console import Console
 from rich.syntax import Syntax
 
@@ -35,6 +36,10 @@ class PipelineComponents(BaseModel):
     """Stores the pipeline components"""
 
     components: list[PipelineComponent] = []
+    graph: nx.DiGraph = Field(default=nx.DiGraph(), exclude=True)
+
+    class Config:
+        arbitrary_types_allowed = True
 
     @property
     def last(self) -> PipelineComponent:
@@ -59,6 +64,10 @@ def __iter__(self) -> Iterator[PipelineComponent]:
     def __len__(self) -> int:
         return len(self.components)
 
+    def validate_graph(self) -> None:
+        if not nx.is_directed_acyclic_graph(self.graph):
+            raise ValueError("Pipeline is not a valid DAG.")
+
     def validate_unique_names(self) -> None:
         step_names = [component.full_name for component in self.components]
         duplicates = [name for name, count in Counter(step_names).items() if count > 1]
@@ -67,6 +76,29 @@ def validate_unique_names(self) -> None:
                 f"step names should be unique. duplicate step names: {', '.join(duplicates)}"
             )
 
+    def generate_graph(self) -> None:
+        for component in self.components:
+            component_node_name = self.__get_vertex_component_name(component)
+            self.graph.add_node(component_node_name)
+
+            for input_topic in component.inputs:
+                self.__add_input(input_topic, component_node_name)
+
+            for output_topic in component.outputs:
+                self.__add_output(output_topic, component_node_name)
+
+    def __add_output(self, output_topic: str, source: str) -> None:
+        self.graph.add_node(output_topic)
+        self.graph.add_edge(source, output_topic)
+
+    def __add_input(self, input_topic: str, component_node_name: str) -> None:
+        self.graph.add_node(input_topic)
+        self.graph.add_edge(input_topic, component_node_name)
+
+    @staticmethod
+    def __get_vertex_component_name(component: PipelineComponent) -> str:
+        return f"component-{component.full_name}"
+
     @staticmethod
     def _populate_component_name(component: PipelineComponent) -> None:  # TODO: remove
         with suppress(
@@ -109,6 +141,7 @@ def __init__(
         self.registry = registry
         self.env_components_index = create_env_components_index(environment_components)
         self.parse_components(component_list)
+        self.components.generate_graph()
         self.validate()
 
     @classmethod
@@ -308,6 +341,7 @@ def substitute_in_component(self, component_as_dict: dict) -> dict:
 
     def validate(self) -> None:
         self.components.validate_unique_names()
+        self.components.validate_graph()
 
     @staticmethod
     def pipeline_filename_environment(path: Path, config: PipelineConfig) -> Path:

pyproject.toml

@@ -38,6 +38,7 @@ cachetools = "^5.2.0"
 dictdiffer = "^0.9.0"
 python-schema-registry-client = "^2.4.1"
 httpx = "^0.24.1"
+networkx = "^3.1"
 
 
 [tool.poetry.group.dev.dependencies]

tests/components/test_producer_app.py

@@ -247,3 +247,37 @@ async def test_should_clean_producer_app_and_deploy_clean_up_job_and_delete_clea
                 False,
             ),
         ]
+
+    def test_get_output_topics(
+        self,
+        config: PipelineConfig,
+        handlers: ComponentHandlers,
+    ):
+        producer_app = ProducerApp(
+            name=self.PRODUCER_APP_NAME,
+            config=config,
+            handlers=handlers,
+            **{
+                "namespace": "test-namespace",
+                "app": {
+                    "namespace": "test-namespace",
+                    "streams": {"brokers": "fake-broker:9092"},
+                },
+                "to": {
+                    "topics": {
+                        "${output_topic_name}": TopicConfig(
+                            type=OutputTopicTypes.OUTPUT, partitions_count=10
+                        ),
+                        "extra-topic-1": TopicConfig(
+                            role="first-extra-topic",
+                            partitions_count=10,
+                        ),
+                    }
+                },
+            },
+        )
+        assert producer_app.get_output_topic() == "${output_topic_name}"
+        assert producer_app.get_extra_output_topics() == {
+            "first-extra-topic": "extra-topic-1"
+        }
+        assert producer_app.get_input_topics() == []

tests/components/test_streams_app.py

@@ -443,3 +443,48 @@ async def test_should_clean_streams_app_and_deploy_clean_up_job_and_delete_clean
                 dry_run,
             ),
         ]
+
+    @pytest.mark.asyncio
+    async def test_get_input_output_topics(
+        self, config: PipelineConfig, handlers: ComponentHandlers
+    ):
+        streams_app = StreamsApp(
+            name=self.STREAMS_APP_NAME,
+            config=config,
+            handlers=handlers,
+            **{
+                "namespace": "test-namespace",
+                "app": {
+                    "streams": {"brokers": "fake-broker:9092"},
+                },
+                "from": {
+                    "topics": {
+                        "example-input": {"type": "input"},
+                        "b": {"type": "input"},
+                        "a": {"type": "input"},
+                        "topic-extra2": {"role": "role2"},
+                        "topic-extra3": {"role": "role2"},
+                        "topic-extra": {"role": "role1"},
+                        ".*": {"type": "pattern"},
+                        "example.*": {
+                            "type": "pattern",
+                            "role": "another-pattern",
+                        },
+                    }
+                },
+                "to": {
+                    "topics": {
+                        "example-output": {"type": "output"},
+                        "extra-topic": {"role": "fake-role"},
+                    }
+                },
+            },
+        )
+
+        assert streams_app.get_input_topics() == ["example-input", "b", "a"]
+        assert streams_app.get_extra_input_topics() == {
+            "role1": ["topic-extra"],
+            "role2": ["topic-extra2", "topic-extra3"],
+        }
+        assert streams_app.get_output_topic() == "example-output"
+        assert streams_app.get_extra_output_topics() == {"fake-role": "extra-topic"}

tests/pipeline/resources/pipeline-with-loop/defaults.yaml

@@ -0,0 +1,19 @@
+pipeline-component:
+  prefix: ""
+
+kubernetes-app:
+  namespace: example-namespace
+
+kafka-connector:
+  namespace: example-namespace
+
+kafka-app:
+  app:
+    streams:
+      brokers: 127.0.0.1:9092
+      schemaRegistryUrl: 127.0.0.1:8081
+
+streams-app:
+  app:
+    labels:
+      pipeline: ${pipeline_name}

tests/pipeline/resources/pipeline-with-loop/pipeline.yaml

@@ -0,0 +1,34 @@
+- type: producer-app
+  name: app1
+  app:
+    image: producer-image
+  to:
+    topics:
+      my-output-topic:
+        type: output
+
+- type: streams-app
+  name: app2
+  app:
+    image: app2-image
+  from:
+    topics:
+      my-output-topic:
+        type: input
+  to:
+    topics:
+      my-app2-topic:
+        type: output
+
+- type: streams-app
+  name: app3
+  app:
+    image: app3-image
+  from:
+    topics:
+      my-app2-topic:
+        type: input
+  to:
+    topics:
+      my-output-topic:
+        type: output

tests/pipeline/resources/pipeline-with-short-topics/defaults.yaml

@@ -9,6 +9,13 @@ kafka-app:
       schema_registry_url: "${schema_registry_url}"
   version: "2.4.2"
 
+producer-app:
+  to:
+    topics:
+      ${output_topic_name}:
+        partitions_count: 3
+
+
 streams-app: # inherits from kafka-app
   app:
     streams:
@@ -19,7 +26,7 @@ streams-app: # inherits from kafka-app
         type:
       error-topic:
         type: error
-      extra-topic:
+      extra-topic-output:
         role: role
   from:
     topics:

tests/pipeline/resources/pipelines-with-graphs/same-topic-and-component-name/config.yaml

@@ -0,0 +1,8 @@
+environment: development
+defaults_path: ..
+brokers: "broker:9092"
+helm_diff_config:
+  enable: false
+kafka_connect_host: "kafka_connect_host:8083"
+kafka_rest_host: "kafka_rest_host:8082"
+