Support networkx to represent components as a graph

kpops/components/base_components/pipeline_component.py

@@ -62,15 +62,18 @@ def __init__(self, **kwargs) -> None:
 
     def get_input_topics(self) -> list[str] | None:
         """Get all the input topics from config."""
+        return []
 
     def get_extra_input_topics(self) -> dict[str, list[str]] | None:
         """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] | None:
         """Get extra output topics list from config."""
+        return {}
 
     @property
     def full_name(self) -> str:
@@ -130,19 +133,18 @@ def set_input_topics(self) -> None:
             for name, topic in self.from_.topics.items():
                 self.apply_from_inputs(name, topic)
 
-    def __get_all_input_topics(self) -> list[str]:
-        input_topics = self.get_input_topics()
-        extra_input_topics = self.get_extra_input_topics()
-        all_input_topics: list[str] = []
-        if input_topics is not None:
-            all_input_topics.extend(input_topics)
-        if extra_input_topics:
-            all_input_topics.extend(chain(*extra_input_topics.values()))
-        return all_input_topics
+    @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 input_topics(self) -> Iterator[str]:
-        yield from self.__get_all_input_topics()
+    def outputs(self) -> Iterator[str]:
+        if self.get_output_topic() is not None:
+            yield self.get_output_topic()
+        yield from self.get_extra_output_topics().values()
+
 
     def __get_all_output_topics(self) -> list[str]:
         all_output_topics: list[str] = []
@@ -154,10 +156,6 @@ def __get_all_output_topics(self) -> list[str]:
             all_output_topics.extend(list(extra_output_topics.values()))
         return all_output_topics
 
-    @property
-    def output_topics(self):
-        yield from self.__get_all_output_topics()
-
     def apply_from_inputs(self, name: str, topic: FromTopic) -> None:
         """Add a `from` section input to the component config
 

kpops/components/streams_bootstrap/streams/streams_app.py

@@ -32,6 +32,10 @@ def get_extra_input_topics(self) -> dict[str, list[str]] | None:
     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

@@ -81,15 +81,15 @@ def generate_graph(self) -> None:
             component_node_name = self.__get_vertex_component_name(component)
             self.graph.add_node(component_node_name)
 
-            self.__add_ingoing_edges(list(component.input_topics), component_node_name)
-            self.__add_output(list(component.output_topics), component_node_name)
+            self.__add_inputs(list(component.inputs), component_node_name)
+            self.__add_outputs(list(component.outputs), component_node_name)
 
-    def __add_output(self, all_output_topics: list[str], source: str) -> None:
+    def __add_outputs(self, all_output_topics: list[str], source: str) -> None:
         for output_topic in all_output_topics:
             self.graph.add_node(output_topic)
             self.graph.add_edge(source, output_topic)
 
-    def __add_ingoing_edges(
+    def __add_inputs(
         self, all_input_topics: list[str], component_node_name: str
     ) -> None:
         for input_topic in all_input_topics:

tests/components/test_producer_app.py

@@ -280,4 +280,4 @@ def test_get_output_topics(
         assert producer_app.get_extra_output_topics() == {
             "first-extra-topic": "extra-topic-1"
         }
-        assert producer_app.get_input_topics() is None
+        assert producer_app.get_input_topics() == []

tests/components/test_streams_app.py

@@ -472,7 +472,14 @@ async def test_get_input_output_topics(
                         },
                     }
                 },
-                "to": {"topics": {"example-output": {"type": "output"}}},
+                "to": {"topics": {
+                    "example-output": {
+                        "type": "output"
+                    },
+                    "extra-topic":{
+                        "role": "fake-role"
+                    }
+                }},
             },
         )
 
@@ -482,3 +489,6 @@ async def test_get_input_output_topics(
             "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-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/test_pipeline.py

@@ -528,10 +528,10 @@ def test_short_topic_definition(self):
         input_components = enriched_pipeline["components"][4]["from"]["components"]
         assert "type" not in output_topics["output-topic"]
         assert output_topics["error-topic"]["type"] == "error"
-        assert "type" not in output_topics["extra-topic"]
+        assert "type" not in output_topics["extra-topic-output"]
         assert "role" not in output_topics["output-topic"]
         assert "role" not in output_topics["error-topic"]
-        assert output_topics["extra-topic"]["role"] == "role"
+        assert output_topics["extra-topic-output"]["role"] == "role"
 
         assert "type" not in ["input-topic"]
         assert "type" not in input_topics["extra-topic"]