Generate schema for pipeline.yaml and config.yaml

gen_schema.py

@@ -0,0 +1,47 @@
+from typing import Annotated, Any, Sequence
+
+from pydantic import Field, schema, schema_json_of
+from pydantic.fields import ModelField
+from pydantic.schema import SkipField
+
+from kpops.components.base_components.kafka_connect import (
+    KafkaSinkConnector,
+    KafkaSourceConnector,
+)
+from kpops.components.streams_bootstrap.producer.producer_app import ProducerApp
+from kpops.components.streams_bootstrap.streams.streams_app import StreamsApp
+
+original_field_schema = schema.field_schema
+
+
+# adapted from https://github.com/tiangolo/fastapi/issues/1378#issuecomment-764966955
+def field_schema(field: ModelField, **kwargs: Any) -> Any:
+    if field.field_info.extra.get("hidden_from_schema"):
+        raise SkipField(f"{field.name} field is being hidden")
+    else:
+        return original_field_schema(field, **kwargs)
+
+
+schema.field_schema = field_schema
+
+ComponentType = (
+    StreamsApp
+    | ProducerApp
+    | KafkaSourceConnector
+    | KafkaSinkConnector
+    # | PipelineComponent
+)
+
+
+AnnotatedPipelineComponent = Annotated[
+    ComponentType, Field(discriminator="schema_type")
+]
+
+schema = schema_json_of(
+    Sequence[AnnotatedPipelineComponent],
+    title="kpops pipeline schema",
+    by_alias=True,
+    indent=4,
+).replace("schema_type", "type")
+
+print(schema)

kpops/cli/registry.py

@@ -29,7 +29,7 @@ def find_components(self, module_name: str) -> None:
         :param module_name: name of the python module
         """
         for _class in _find_classes(module_name, PipelineComponent):
-            self._classes[_class._type] = _class
+            self._classes[_class.type] = _class
 
     def __getitem__(self, component_type: str) -> type[PipelineComponent]:
         try:

kpops/components/base_components/base_defaults_component.py

@@ -4,7 +4,7 @@
 from collections import deque
 from collections.abc import Mapping, Sequence
 from pathlib import Path
-from typing import TypeVar
+from typing import Any, ClassVar, TypeVar
 
 import typer
 from pydantic import BaseConfig, BaseModel, Field
@@ -17,21 +17,23 @@
 
 
 class BaseDefaultsComponent(BaseModel):
-    _type: str = Field(..., alias="type")
+    type: ClassVar[str] = Field(default=..., const=True)
 
-    enrich: bool = Field(default=False, exclude=True)
-    config: PipelineConfig = Field(default=..., exclude=True)
-    handlers: ComponentHandlers = Field(default=..., exclude=True)
+    enrich: bool = Field(default=False, exclude=True, hidden_from_schema=True)
+    config: PipelineConfig = Field(default=..., exclude=True, hidden_from_schema=True)
+    handlers: ComponentHandlers = Field(
+        default=..., exclude=True, hidden_from_schema=True
+    )
 
     class Config(BaseConfig):
         arbitrary_types_allowed = True
 
-    def __init__(self, **kwargs):
+    def __init__(self, **kwargs) -> None:
         if kwargs.get("enrich", True):
             kwargs = self.extend_with_defaults(kwargs)
         super().__init__(**kwargs)
 
-    def extend_with_defaults(self, kwargs) -> dict:
+    def extend_with_defaults(self, kwargs: dict[str, Any]) -> dict:
         """
         Merges tmp_defaults with all tmp_defaults for parent classes
 
@@ -60,13 +62,13 @@ def extend_with_defaults(self, kwargs) -> dict:
                 if not environment_default_file_path.exists():
                     kwargs = update_nested(
                         kwargs,
-                        defaults_from_yaml(main_default_file_path, base._type),
+                        defaults_from_yaml(main_default_file_path, base.type),
                     )
                 else:
                     kwargs = update_nested(
                         kwargs,
-                        defaults_from_yaml(environment_default_file_path, base._type),
-                        defaults_from_yaml(main_default_file_path, base._type),
+                        defaults_from_yaml(environment_default_file_path, base.type),
+                        defaults_from_yaml(main_default_file_path, base.type),
                     )
         return kwargs
 

kpops/components/base_components/kafka_app.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from typing import ClassVar
 
 from pydantic import BaseModel, Extra
 from typing_extensions import override
@@ -37,7 +38,7 @@ class KafkaApp(KubernetesApp):
     Producer or streaming apps should inherit from this class.
     """
 
-    _type = "kafka-app"
+    type: ClassVar[str] = "kafka-app"
     app: KafkaAppConfig
 
     def __init__(self, **kwargs) -> None:

kpops/components/base_components/kafka_connect.py

@@ -1,8 +1,9 @@
 import json
 import os
 from abc import ABC
-from typing import NoReturn
+from typing import ClassVar, Literal, NoReturn
 
+from pydantic import Field
 from typing_extensions import override
 
 from kpops.cli.pipeline_config import ENV_PREFIX
@@ -14,7 +15,7 @@
 
 
 class KafkaConnector(PipelineComponent, ABC):
-    _type = "kafka-connect"
+    type: ClassVar[str] = "kafka-connect"
     app: KafkaConnectConfig
 
     def __init__(self, **kwargs) -> None:
@@ -64,7 +65,10 @@ def clean(self, dry_run: bool) -> None:
 
 
 class KafkaSourceConnector(KafkaConnector):
-    _type = "kafka-source-connector"
+    type: ClassVar[str] = "kafka-source-connector"
+    schema_type: Literal["kafka-source-connector"] = Field(
+        default="kafka-source-connector", exclude=True
+    )
 
     @override
     def apply_from_inputs(self, name: str, topic: FromTopic) -> NoReturn:
@@ -90,7 +94,10 @@ def __run_kafka_connect_resetter(self, dry_run: bool) -> None:
 
 
 class KafkaSinkConnector(KafkaConnector):
-    _type = "kafka-sink-connector"
+    type: ClassVar[str] = "kafka-sink-connector"
+    schema_type: Literal["kafka-sink-connector"] = Field(
+        default="kafka-sink-connector", exclude=True
+    )
 
     @override
     def add_input_topics(self, topics: list[str]) -> None:

kpops/components/base_components/kubernetes_app.py

@@ -3,6 +3,7 @@
 import logging
 import re
 from functools import cached_property
+from typing import ClassVar
 
 from pydantic import BaseModel
 from typing_extensions import override
@@ -35,7 +36,7 @@ class Config(CamelCaseConfig):
 class KubernetesApp(PipelineComponent):
     """Base kubernetes app"""
 
-    _type = "kubernetes-app"
+    type: ClassVar[str] = "kubernetes-app"
     app: KubernetesAppConfig
 
     version: str | None = None

kpops/components/base_components/pipeline_component.py

@@ -60,17 +60,17 @@ def substitute_component_variables(self, topic_name: str) -> str:
         """
         error_topic_name = self.substitute_component_names(
             self.config.topic_name_config.default_error_topic_name,
-            self._type,
+            self.type,
             **os.environ,
         )
         output_topic_name = self.substitute_component_names(
             self.config.topic_name_config.default_output_topic_name,
-            self._type,
+            self.type,
             **os.environ,
         )
         return self.substitute_component_names(
             topic_name,
-            self._type,
+            self.type,
             error_topic_name=error_topic_name,
             output_topic_name=output_topic_name,
         )
@@ -147,7 +147,7 @@ def weave_from_topics(self, prev_component_to: ToSection) -> None:
 
     def substitute_name(self):
         if self.name:
-            self.name = self.substitute_component_names(self.name, self._type)
+            self.name = self.substitute_component_names(self.name, self.type)
         else:
             raise ValueError("Every component must have a name in the end.")
 

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -1,4 +1,6 @@
-from pydantic import BaseConfig, Extra
+from typing import ClassVar, Literal
+
+from pydantic import BaseConfig, Extra, Field
 from typing_extensions import override
 
 from kpops.component_handlers.helm_wrapper.model import HelmRepoConfig
@@ -18,7 +20,8 @@ class ProducerApp(KafkaApp):
     This producer holds configuration to use as values for the streams bootstrap produce helm chart.
     """
 
-    _type = "producer"
+    type: ClassVar[str] = "producer"
+    schema_type: Literal["producer"] = Field(default="producer", exclude=True)
     app: ProducerValues
 
     class Config(BaseConfig):

kpops/components/streams_bootstrap/streams/streams_app.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from pydantic import BaseConfig, Extra
+from typing import ClassVar, Literal
+
+from pydantic import BaseConfig, Extra, Field
 from typing_extensions import override
 
 from kpops.component_handlers.helm_wrapper.model import HelmRepoConfig
@@ -14,7 +16,8 @@ class StreamsApp(KafkaApp):
     StreamsApp component that configures a streams bootstrap app
     """
 
-    _type = "streams-app"
+    type: ClassVar[str] = "streams-app"
+    schema_type: Literal["streams-app"] = Field(default="streams-app", exclude=True)
     app: StreamsAppConfig
 
     class Config(BaseConfig):

kpops/pipeline_generator/pipeline.py

@@ -227,7 +227,7 @@ def substitute_component_specific_variables(
             substitute(
                 json.dumps(pair),
                 {
-                    "component_type": component_object._type,
+                    "component_type": component_object.type,
                     "component_name": component_object.name,
                 },
             )