Refactor enrichment using Pydantic model validator (#444)

kpops/components/base_components/base_defaults_component.py

@@ -75,12 +75,18 @@ class BaseDefaultsComponent(DescConfigModel, ABC):
         exclude=True,
     )
 
-    def __init__(self, **kwargs) -> None:
-        if kwargs.get("enrich", True):
-            kwargs = self.extend_with_defaults(**kwargs)
-        super().__init__(**kwargs)
-        if kwargs.get("validate", True):
-            self._validate_custom(**kwargs)
+    @pydantic.model_validator(mode="before")
+    @classmethod
+    def enrich_component(cls, values: dict[str, Any]) -> dict[str, Any]:
+        if values.get("enrich", True):
+            values = cls.extend_with_defaults(**values)
+        return values
+
+    @pydantic.model_validator(mode="after")
+    def validate_component(self) -> Self:
+        if self.validate_:
+            self._validate_custom()
+        return self
 
     @computed_field
     @cached_classproperty
@@ -156,11 +162,8 @@ def load_defaults(cls, *defaults_file_paths: Path) -> dict[str, Any]:
             )
         return defaults
 
-    def _validate_custom(self, **kwargs) -> None:
-        """Run custom validation on component.
-
-        :param kwargs: The init kwargs for the component
-        """
+    def _validate_custom(self) -> None:
+        """Run custom validation on component."""
 
 
 def defaults_from_yaml(path: Path, key: str) -> dict:

kpops/components/base_components/kubernetes_app.py

@@ -45,8 +45,8 @@ class KubernetesApp(PipelineComponent, ABC):
     )
 
     @override
-    def _validate_custom(self, **kwargs) -> None:
-        super()._validate_custom(**kwargs)
+    def _validate_custom(self) -> None:
+        super()._validate_custom()
         self.validate_kubernetes_name(self.name)
 
     @staticmethod

kpops/utils/gen_schema.py

@@ -137,7 +137,10 @@ def gen_pipeline_schema(
             default=component.type,
         )
         core_schema: DefinitionsSchema = component.__pydantic_core_schema__  # pyright:ignore[reportGeneralTypeIssues]
-        model_schema: ModelFieldsSchema = core_schema["schema"]["schema"]  # pyright:ignore[reportGeneralTypeIssues,reportTypedDictNotRequiredAccess]
+
+        model_schema: ModelFieldsSchema = core_schema["schema"]["schema"]["schema"][  # pyright:ignore[reportGeneralTypeIssues,reportTypedDictNotRequiredAccess]
+            "schema"
+        ]
         model_schema["fields"]["type"] = ModelField(
             type="model-field",
             schema=LiteralSchema(