Fix substitution

docs/docs/schema/defaults.json

@@ -586,7 +586,7 @@
                     "title": "Brokers",
                     "type": "string"
                 },
-                "schemaRegistryUrl": {
+                "schema_registry_url": {
                     "anyOf": [
                         {
                             "type": "string"
@@ -597,7 +597,7 @@
                     ],
                     "default": null,
                     "description": "URL of the schema registry",
-                    "title": "Schemaregistryurl"
+                    "title": "Schema Registry Url"
                 }
             },
             "required": [
@@ -886,7 +886,7 @@
                     "description": "Output topic",
                     "title": "Outputtopic"
                 },
-                "schemaRegistryUrl": {
+                "schema_registry_url": {
                     "anyOf": [
                         {
                             "type": "string"
@@ -897,7 +897,7 @@
                     ],
                     "default": null,
                     "description": "URL of the schema registry",
-                    "title": "Schemaregistryurl"
+                    "title": "Schema Registry Url"
                 }
             },
             "required": [
@@ -1396,7 +1396,7 @@
                     "description": "Output topic",
                     "title": "Outputtopic"
                 },
-                "schemaRegistryUrl": {
+                "schema_registry_url": {
                     "anyOf": [
                         {
                             "type": "string"
@@ -1407,7 +1407,7 @@
                     ],
                     "default": null,
                     "description": "URL of the schema registry",
-                    "title": "Schemaregistryurl"
+                    "title": "Schema Registry Url"
                 }
             },
             "required": [

docs/docs/schema/pipeline.json

@@ -576,7 +576,7 @@
                     "description": "Output topic",
                     "title": "Outputtopic"
                 },
-                "schemaRegistryUrl": {
+                "schema_registry_url": {
                     "anyOf": [
                         {
                             "type": "string"
@@ -587,7 +587,7 @@
                     ],
                     "default": null,
                     "description": "URL of the schema registry",
-                    "title": "Schemaregistryurl"
+                    "title": "Schema Registry Url"
                 }
             },
             "required": [
@@ -992,7 +992,7 @@
                     "description": "Output topic",
                     "title": "Outputtopic"
                 },
-                "schemaRegistryUrl": {
+                "schema_registry_url": {
                     "anyOf": [
                         {
                             "type": "string"
@@ -1003,7 +1003,7 @@
                     ],
                     "default": null,
                     "description": "URL of the schema registry",
-                    "title": "Schemaregistryurl"
+                    "title": "Schema Registry Url"
                 }
             },
             "required": [

kpops/cli/main.py

@@ -440,7 +440,7 @@ async def async_reset():
             pipeline_tasks = pipeline.build_execution_graph(reset_runner, reverse=True)
             await pipeline_tasks
         else:
-            for component in pipeline.components:
+            for component in reversed(pipeline.components):
                 await reset_runner(component)
 
     asyncio.run(async_reset())
@@ -481,7 +481,7 @@ async def async_clean():
             pipeline_tasks = pipeline.build_execution_graph(clean_runner, reverse=True)
             await pipeline_tasks
         else:
-            for component in pipeline.components:
+            for component in reversed(pipeline.components):
                 await clean_runner(component)
 
     asyncio.run(async_clean())

kpops/components/base_components/base_defaults_component.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import logging
 from abc import ABC
 from collections.abc import Sequence
@@ -21,11 +22,16 @@
 from kpops.component_handlers import ComponentHandlers
 from kpops.config import KpopsConfig
 from kpops.utils import cached_classproperty
-from kpops.utils.dict_ops import update_nested, update_nested_pair
+from kpops.utils.dict_ops import (
+    generate_substitution,
+    update_nested,
+    update_nested_pair,
+)
 from kpops.utils.docstring import describe_attr
 from kpops.utils.environment import ENV
 from kpops.utils.pydantic import DescConfigModel, issubclass_patched, to_dash
-from kpops.utils.yaml import load_yaml_file
+from kpops.utils.types import JsonType
+from kpops.utils.yaml import load_yaml_file, substitute_nested
 
 try:
     from typing import Self
@@ -54,7 +60,7 @@ class BaseDefaultsComponent(DescConfigModel, ABC):
     )
 
     enrich: SkipJsonSchema[bool] = Field(
-        default=False,
+        default=True,
         description=describe_attr("enrich", __doc__),
         exclude=True,
     )
@@ -70,21 +76,33 @@ class BaseDefaultsComponent(DescConfigModel, ABC):
     )
     validate_: SkipJsonSchema[bool] = Field(
         validation_alias=AliasChoices("validate", "validate_"),
-        default=True,
+        default=False,
         description=describe_attr("validate", __doc__),
         exclude=True,
     )
 
-    @pydantic.model_validator(mode="before")
-    @classmethod
-    def enrich_component(cls, values: dict[str, Any]) -> dict[str, Any]:
+    def __init__(self, **values: Any) -> None:
         if values.get("enrich", True):
+            cls = self.__class__
             values = cls.extend_with_defaults(**values)
-        return values
+            tmp_self = cls(**values, enrich=False)
+            values = tmp_self.model_dump(mode="json", by_alias=True)
+            values = cls.substitute_in_component(tmp_self.config, **values)
+            # HACK: why is double substitution necessary for test_substitute_in_component
+            values = cls.substitute_in_component(tmp_self.config, **values)
+            self.__init__(
+                enrich=False,
+                validate=True,
+                config=tmp_self.config,
+                handlers=tmp_self.handlers,
+                **values,
+            )
+        else:
+            super().__init__(**values)
 
     @pydantic.model_validator(mode="after")
     def validate_component(self) -> Self:
-        if self.validate_:
+        if not self.enrich and self.validate_:
             self._validate_custom()
         return self
 
@@ -113,6 +131,42 @@ def gen_parents():
 
         return tuple(gen_parents())
 
+    @classmethod
+    def substitute_in_component(
+        cls, config: KpopsConfig, **component_data: Any
+    ) -> dict[str, Any]:
+        """Substitute all $-placeholders in a component in dict representation.
+
+        :param component_as_dict: Component represented as dict
+        :return: Updated component
+        """
+        # Leftover variables that were previously introduced in the component by the substitution
+        # functions, still hardcoded, because of their names.
+        # TODO(Ivan Yordanov): Get rid of them
+        substitution_hardcoded: dict[str, JsonType] = {
+            "error_topic_name": config.topic_name_config.default_error_topic_name,
+            "output_topic_name": config.topic_name_config.default_output_topic_name,
+        }
+        component_substitution = generate_substitution(
+            component_data,
+            "component",
+            substitution_hardcoded,
+            separator=".",
+        )
+        substitution = generate_substitution(
+            config.model_dump(mode="json"),
+            "config",
+            existing_substitution=component_substitution,
+            separator=".",
+        )
+
+        return json.loads(
+            substitute_nested(
+                json.dumps(component_data),
+                **update_nested_pair(substitution, ENV),
+            )
+        )
+
     @classmethod
     def extend_with_defaults(cls, config: KpopsConfig, **kwargs: Any) -> dict[str, Any]:
         """Merge parent components' defaults with own.

kpops/components/base_components/kafka_app.py

@@ -3,7 +3,7 @@
 import logging
 from abc import ABC
 
-from pydantic import ConfigDict, Field
+from pydantic import AliasChoices, ConfigDict, Field
 from typing_extensions import override
 
 from kpops.component_handlers.helm_wrapper.model import (
@@ -28,7 +28,11 @@ class KafkaStreamsConfig(CamelCaseConfigModel, DescConfigModel):
 
     brokers: str = Field(default=..., description=describe_attr("brokers", __doc__))
     schema_registry_url: str | None = Field(
-        default=None, description=describe_attr("schema_registry_url", __doc__)
+        default=None,
+        validation_alias=AliasChoices(
+            "schema_registry_url", "schemaRegistryUrl"
+        ),  # TODO: same for other camelcase fields, avoids duplicates during enrichment
+        description=describe_attr("schema_registry_url", __doc__),
     )
 
     model_config = ConfigDict(

kpops/components/base_components/kafka_connector.py

@@ -148,7 +148,6 @@ def connector_config_should_have_component_name(
         app["name"] = component_name
         return KafkaConnectorConfig(**app)
 
-    @computed_field
     @cached_property
     def _resetter(self) -> KafkaConnectorResetter:
         kwargs: dict[str, Any] = {}
@@ -218,6 +217,11 @@ class KafkaSourceConnector(KafkaConnector):
 
     _connector_type: KafkaConnectorType = PrivateAttr(KafkaConnectorType.SOURCE)
 
+    @computed_field
+    @cached_property
+    def _resetter(self) -> KafkaConnectorResetter:
+        return super()._resetter
+
     @override
     def apply_from_inputs(self, name: str, topic: FromTopic) -> NoReturn:
         msg = "Kafka source connector doesn't support FromSection"
@@ -240,6 +244,11 @@ class KafkaSinkConnector(KafkaConnector):
 
     _connector_type: KafkaConnectorType = PrivateAttr(KafkaConnectorType.SINK)
 
+    @computed_field
+    @cached_property
+    def _resetter(self) -> KafkaConnectorResetter:
+        return super()._resetter
+
     @property
     @override
     def input_topics(self) -> list[str]:

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -1,6 +1,6 @@
 from functools import cached_property
 
-from pydantic import Field
+from pydantic import Field, computed_field
 from typing_extensions import override
 
 from kpops.components.base_components.kafka_app import (
@@ -51,12 +51,13 @@ class ProducerApp(KafkaApp, StreamsBootstrap):
         description=describe_attr("from_", __doc__),
     )
 
+    @computed_field
     @cached_property
     def _cleaner(self) -> ProducerAppCleaner:
         return ProducerAppCleaner(
             config=self.config,
             handlers=self.handlers,
-            **self.model_dump(),
+            **self.model_dump(exclude={"_cleaner"}),
         )
 
     @override

kpops/components/streams_bootstrap/streams/streams_app.py

@@ -1,6 +1,6 @@
 from functools import cached_property
 
-from pydantic import Field
+from pydantic import Field, computed_field
 from typing_extensions import override
 
 from kpops.components.base_components.kafka_app import (
@@ -33,12 +33,13 @@ class StreamsApp(KafkaApp, StreamsBootstrap):
         description=describe_attr("app", __doc__),
     )
 
+    @computed_field
     @cached_property
     def _cleaner(self) -> StreamsAppCleaner:
         return StreamsAppCleaner(
             config=self.config,
             handlers=self.handlers,
-            **self.model_dump(),
+            **self.model_dump(exclude={"_cleaner"}),
         )
 
     @property