Support multiple inheritance for doc generation

docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml

@@ -79,12 +79,57 @@ kpops_components_fields:
   - repo_config
   - version
 kpops_components_inheritance_ref:
-  helm-app: kubernetes-app
-  kafka-app: helm-app
-  kafka-connector: pipeline-component
-  kafka-sink-connector: kafka-connector
-  kafka-source-connector: kafka-connector
-  kubernetes-app: pipeline-component
-  pipeline-component: base-defaults-component
-  producer-app: kafka-app
-  streams-app: kafka-app
+  helm-app:
+    bases:
+    - kubernetes-app
+    parents:
+    - kubernetes-app
+    - pipeline-component
+  kafka-app:
+    bases:
+    - helm-app
+    parents:
+    - helm-app
+    - kubernetes-app
+    - pipeline-component
+  kafka-connector:
+    bases:
+    - pipeline-component
+    parents:
+    - pipeline-component
+  kafka-sink-connector:
+    bases:
+    - kafka-connector
+    parents:
+    - kafka-connector
+    - pipeline-component
+  kafka-source-connector:
+    bases:
+    - kafka-connector
+    parents:
+    - kafka-connector
+    - pipeline-component
+  kubernetes-app:
+    bases:
+    - pipeline-component
+    parents:
+    - pipeline-component
+  pipeline-component:
+    bases: []
+    parents: []
+  producer-app:
+    bases:
+    - kafka-app
+    parents:
+    - kafka-app
+    - helm-app
+    - kubernetes-app
+    - pipeline-component
+  streams-app:
+    bases:
+    - kafka-app
+    parents:
+    - kafka-app
+    - helm-app
+    - kubernetes-app
+    - pipeline-component

hooks/gen_docs/gen_docs_components.py

@@ -11,6 +11,7 @@
 from kpops.cli.registry import _find_classes
 from kpops.components import KafkaConnector, PipelineComponent
 from kpops.utils.colorify import redify, yellowify
+from kpops.utils.pydantic import issubclass_patched
 from kpops.utils.yaml import load_yaml_file
 
 PATH_KPOPS_MAIN = ROOT / "kpops/cli/main.py"
@@ -33,14 +34,6 @@
 )
 
 KPOPS_COMPONENTS = tuple(_find_classes("kpops.components", PipelineComponent))
-KPOPS_COMPONENTS_INHERITANCE_REF = {
-    component.type: cast(
-        type[PipelineComponent],
-        component.__base__,
-    ).type
-    for component in KPOPS_COMPONENTS
-}
-
 KPOPS_COMPONENTS_SECTIONS = {
     component.type: [
         field_name
@@ -49,6 +42,27 @@
     ]
     for component in KPOPS_COMPONENTS
 }
+KPOPS_COMPONENTS_INHERITANCE_REF = {
+    component.type: {
+        "bases": [
+            cast(
+                type[PipelineComponent],
+                base,
+            ).type
+            for base in component.__bases__
+            if issubclass_patched(base, PipelineComponent)
+        ],
+        "parents": [
+            cast(
+                type[PipelineComponent],
+                parent,
+            ).type
+            for parent in component.parents
+        ],
+    }
+    for component in KPOPS_COMPONENTS
+}
+
 # Dependency files should not be changed manually
 DANGEROUS_FILES_TO_CHANGE = {
     PATH_DOCS_COMPONENTS_DEPENDENCIES,
@@ -92,14 +106,13 @@ def filter_sections(
         if section := filter_section(component_name, sections, target_section):
             component_sections.append(section)
         elif include_inherited:
-            temp_component_name = component_name
-            while (
-                temp_component_name := KPOPS_COMPONENTS_INHERITANCE_REF[
-                    temp_component_name
-                ]
-            ) != PipelineComponent.type:
+            for component in KPOPS_COMPONENTS_INHERITANCE_REF[component_name][
+                "parents"
+            ]:
+                if component == PipelineComponent.type:
+                    break
                 if section := filter_section(
-                    temp_component_name,
+                    component,
                     sections,
                     target_section,
                 ):
@@ -123,11 +136,12 @@ def filter_section(
     section = target_section + "-" + component_name + ".yaml"
     if section in sections:
         return section
-    if KPOPS_COMPONENTS_INHERITANCE_REF[component_name] == PipelineComponent.type:
+    if KPOPS_COMPONENTS_INHERITANCE_REF[component_name]["bases"] == [
+        PipelineComponent.type
+    ]:
         section = target_section + ".yaml"
         if section in sections:
             return section
-        return None
     return None
 
 

hooks/gen_docs/gen_docs_env_vars.py

@@ -25,6 +25,7 @@
 from hooks.gen_docs import IterableStrEnum
 from kpops.cli import main
 from kpops.config import KpopsConfig
+from kpops.utils.pydantic import issubclass_patched
 
 PATH_DOCS_RESOURCES = ROOT / "docs/docs/resources"
 PATH_DOCS_VARIABLES = PATH_DOCS_RESOURCES / "variables"
@@ -284,29 +285,9 @@ def collect_fields(model: type[BaseModel]) -> dict[str, Any]:
     :param model: settings class
     :return: ``dict`` of all fields in a settings class
     """
-
-    def patched_issubclass_of_basemodel(cls):
-        """Pydantic breaks issubclass.
-
-        ``issubclass(set[str], set)  # True``
-        ``issubclass(BaseSettings, BaseModel)  # True``
-        ``issubclass(set[str], BaseModel)  # raises exception``
-
-        :param cls: class to check
-        :return: Whether cls is subclass of ``BaseModel``
-        """
-        try:
-            return issubclass(cls, BaseModel)
-        except TypeError as e:
-            if str(e) == "issubclass() arg 1 must be a class":
-                return False
-            raise
-
     seen_fields = {}
     for field_name, field_value in model.model_fields.items():
-        if field_value.annotation and patched_issubclass_of_basemodel(
-            field_value.annotation
-        ):
+        if field_value.annotation and issubclass_patched(field_value.annotation):
             seen_fields[field_name] = collect_fields(field_value.annotation)
         else:
             seen_fields[field_name] = field_value

kpops/components/base_components/pipeline_component.py

@@ -18,7 +18,14 @@
     TopicConfig,
     ToSection,
 )
+from kpops.utils import cached_classproperty
 from kpops.utils.docstring import describe_attr
+from kpops.utils.pydantic import issubclass_patched
+
+try:
+    from typing import Self
+except ImportError:
+    from typing_extensions import Self
 
 
 class PipelineComponent(BaseDefaultsComponent, ABC):
@@ -64,6 +71,22 @@ def __init__(self, **kwargs) -> None:
     def full_name(self) -> str:
         return self.prefix + self.name
 
+    @cached_classproperty
+    def parents(cls: type[Self]) -> tuple[type[PipelineComponent], ...]:  # pyright: ignore[reportGeneralTypeIssues]
+        """Get parent components.
+
+        :return: All ancestor KPOps components
+        """
+
+        def gen_parents():
+            for base in cls.mro():
+                # skip class itself and non-component ancestors
+                if base is cls or not issubclass_patched(base, PipelineComponent):
+                    continue
+                yield base
+
+        return tuple(gen_parents())
+
     def add_input_topics(self, topics: list[str]) -> None:
         """Add given topics to the list of input topics.
 

kpops/utils/pydantic.py

@@ -95,6 +95,27 @@ def exclude_defaults(model: BaseModel, dumped_model: dict[str, _V]) -> dict[str,
     }
 
 
+def issubclass_patched(
+    __cls: type, __class_or_tuple: type | tuple[type, ...] = BaseModel
+) -> bool:
+    """Pydantic breaks ``issubclass``.
+
+    ``issubclass(set[str], set)  # True``
+    ``issubclass(BaseSettings, BaseModel)  # True``
+    ``issubclass(set[str], BaseModel)  # raises exception``
+
+    :param cls: class to check
+    :base: class(es) to check against, defaults to ``BaseModel``
+    :return: Whether 'cls' is derived from another class or is the same class.
+    """
+    try:
+        return issubclass(__cls, __class_or_tuple)
+    except TypeError as e:
+        if str(e) == "issubclass() arg 1 must be a class":
+            return False
+        raise
+
+
 class CamelCaseConfigModel(BaseModel):
     model_config = ConfigDict(
         alias_generator=to_camel,