Support networkx to represent components as a graph

kpops/components/base_components/kafka_connector.py

@@ -411,6 +411,11 @@ class KafkaSinkConnector(KafkaConnector):
         exclude=True,
     )
 
+    @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

@@ -75,6 +75,20 @@ def __init__(self, **kwargs) -> None:
         self.set_input_topics()
         self.set_output_topics()
 
+    def get_input_topics(self) -> list[str] | None:
+        """
+        Get all the input topics
+        """
+
+    def get_extra_output_topics(self) -> dict[str, str] | None:
+        ...
+
+    def get_extra_input_topics(self) -> dict[str, list[str]] | None:
+        ...
+
+    def get_output_topic(self) -> str | None:
+        ...
+
     def add_input_topics(self, topics: list[str]) -> None:
         """Add given topics to the list of input topics.
 

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -63,6 +63,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

@@ -42,6 +42,18 @@ class StreamsApp(KafkaApp):
     class Config(DescConfig):
         extra = Extra.allow
 
+    @override
+    def get_input_topics(self) -> list[str] | None:
+        return self.app.streams.input_topics
+
+    @override
+    def get_extra_input_topics(self) -> dict[str, list[str]] | None:
+        return self.app.streams.extra_input_topics
+
+    @override
+    def get_output_topic(self) -> str | None:
+        return self.app.streams.output_topic
+
     @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_components: 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_components(self) -> None:
+        if not nx.is_directed_acyclic_graph(self.graph_components):
+            raise ValueError("Component graph contain loops!")
+
     def validate_unique_names(self) -> None:
         step_names = [component.name for component in self.components]
         duplicates = [name for name, count in Counter(step_names).items() if count > 1]
@@ -112,6 +121,7 @@ def __init__(
         self.registry = registry
         self.env_components_index = create_env_components_index(environment_components)
         self.parse_components(component_list)
+        self.__generate_graph()
         self.validate()
 
     @classmethod
@@ -160,6 +170,59 @@ def load_from_yaml(
         pipeline = cls(main_content, env_content, registry, config, handlers)
         return pipeline
 
+    def __generate_graph(self):
+        for component in self.components:
+            all_input_topics = self.__get_all_input_topics(component)
+            all_output_topics = self.__get_all_output_topics(component)
+
+            component_node_name = self.__get_vertex_component_name(component)
+            self.components.graph_components.add_node(component_node_name)
+
+            self.__add_ingoing_edges(all_input_topics, component_node_name)
+            self.__add_outgoing_edges(all_output_topics, component_node_name)
+
+    def __add_outgoing_edges(
+        self, all_output_topics: list[str], component_node_name: str
+    ) -> None:
+        for output_topic in all_output_topics:
+            self.components.graph_components.add_node(output_topic)
+            self.components.graph_components.add_edge(component_node_name, output_topic)
+
+    def __add_ingoing_edges(
+        self, all_input_topics: list[str], component_node_name: str
+    ) -> None:
+        for input_topic in all_input_topics:
+            self.components.graph_components.add_node(input_topic)
+            self.components.graph_components.add_edge(input_topic, component_node_name)
+
+    def __get_vertex_component_name(self, component: PipelineComponent) -> str:
+        component_vertex_name = f"component-{component.name}"
+        return component_vertex_name
+
+    def __get_all_output_topics(self, component: PipelineComponent) -> list[str]:
+        all_output_topics: list[str] = []
+        output_topics = component.get_output_topic()
+        extra_output_topics = component.get_extra_output_topics()
+        if output_topics is not None:
+            all_output_topics += [output_topics]
+        if extra_output_topics is not None and extra_output_topics:
+            all_output_topics += list(extra_output_topics.values())
+        return all_output_topics
+
+    def __get_all_input_topics(self, component: PipelineComponent) -> list[str]:
+        input_topics = component.get_input_topics()
+        extra_input_topics = component.get_extra_input_topics()
+        all_input_topics: list[str] = []
+        if input_topics is not None:
+            all_input_topics += input_topics
+        if extra_input_topics is not None and extra_input_topics:
+            all_input_topics += [
+                topic
+                for list_topics in extra_input_topics.values()
+                for topic in list_topics
+            ]
+        return all_input_topics
+
     def parse_components(self, component_list: list[dict]) -> None:
         """Instantiate, enrich and inflate a list of components
 
@@ -327,6 +390,7 @@ def substitute_in_component(self, component_as_dict: dict) -> dict:
 
     def validate(self) -> None:
         self.components.validate_unique_names()
+        self.components.validate_graph_components()
 
     @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

@@ -240,3 +240,36 @@ async def test_should_clean_producer_app_and_deploy_clean_up_job_and_delete_clea
                 "test-namespace", self.PRODUCER_APP_CLEAN_NAME, 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(
+                            type=OutputTopicTypes.EXTRA,
+                            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"
+        }

tests/components/test_streams_app.py

@@ -429,3 +429,62 @@ async def test_should_clean_streams_app_and_deploy_clean_up_job_and_delete_clean
                 "test-namespace", self.STREAMS_APP_CLEAN_NAME, dry_run
             ),
         ]
+
+    @pytest.mark.asyncio
+    async def test_get_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": {"type": "extra", "role": "role2"},
+                        "topic-extra3": {"type": "extra", "role": "role2"},
+                        "topic-extra": {"type": "extra", "role": "role1"},
+                        ".*": {"type": "input-pattern"},
+                        "example.*": {
+                            "type": "extra-pattern",
+                            "role": "another-pattern",
+                        },
+                    }
+                },
+            },
+        )
+        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"],
+        }
+
+    @pytest.mark.asyncio
+    async def test_get_output_topic(
+        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"},
+                    }
+                },
+                "to": {"topics": {"example-output": {"type": "output"}}},
+            },
+        )
+        assert streams_app.get_output_topic() == "example-output"

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

@@ -0,0 +1,22 @@
+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
+      optimizeLeaveGroupBehavior: false
+
+streams-app:
+  app:
+    labels:
+      pipeline: ${pipeline_name}
+    streams:
+      optimizeLeaveGroupBehavior: false

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

@@ -0,0 +1,34 @@
+- type: producer
+  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/test_pipeline.py

@@ -480,3 +480,18 @@ def test_validate_unique_step_names(self):
                 ],
                 catch_exceptions=False,
             )
+
+    def test_validate_loops_on_pipeline(self):
+        with pytest.raises(ValueError, match="Component graph contain loops!"):
+            runner.invoke(
+                app,
+                [
+                    "generate",
+                    "--pipeline-base-dir",
+                    str(PIPELINE_BASE_DIR_PATH),
+                    str(RESOURCE_PATH / "pipeline-with-loop/pipeline.yaml"),
+                    "--defaults",
+                    str(RESOURCE_PATH / "pipeline-with-loop"),
+                ],
+                catch_exceptions=False,
+            )