implement config transformations: AddFields and RemoveFields`

Author: pnilan

airbyte_cdk/sources/declarative/transformations/config_transformations/__init__.py

@@ -2,6 +2,8 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
+from .add_fields import AddFields
 from .remap_field import RemapField
+from .remove_fields import RemoveFields
 
-__all__ = ["RemapField"]
+__all__ = ["RemapField", "AddFields", "RemoveFields"]

airbyte_cdk/sources/declarative/transformations/config_transformations/add_fields.py

@@ -0,0 +1,120 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass, field
+from typing import Any, List, MutableMapping, Optional, Type, Union
+
+import dpath
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.sources.declarative.transformations.config_transformations.config_transformation import (
+    ConfigTransformation,
+)
+from airbyte_cdk.sources.types import FieldPointer
+
+
+@dataclass(frozen=True)
+class AddedFieldDefinition:
+    """Defines the field to add on a config"""
+
+    path: FieldPointer
+    value: Union[InterpolatedString, str]
+    value_type: Optional[Type[Any]] = None
+
+
+@dataclass(frozen=True)
+class ParsedAddFieldDefinition:
+    """Defines the field to add on a config"""
+
+    path: FieldPointer
+    value: InterpolatedString
+    value_type: Optional[Type[Any]] = None
+
+
+@dataclass
+class AddFields(ConfigTransformation):
+    """
+    Transformation which adds fields to a config. The path of the added field can be nested. Adding nested fields will create all
+    necessary parent objects (like mkdir -p).
+
+    This transformation has access to the config being transformed.
+
+    Examples of instantiating this transformation via YAML:
+    - type: AddFields
+      fields:
+        # hardcoded constant
+        - path: ["path"]
+          value: "static_value"
+
+        # nested path
+        - path: ["path", "to", "field"]
+          value: "static"
+
+        # from config
+        - path: ["derived_field"]
+          value: "{{ config.original_field }}"
+
+        # by supplying any valid Jinja template directive or expression
+        - path: ["two_times_two"]
+          value: "{{ 2 * 2 }}"
+
+    Attributes:
+        fields (List[AddedFieldDefinition]): A list of transformations (path and corresponding value) that will be added to the config
+    """
+
+    fields: List[AddedFieldDefinition]
+    condition: str = ""
+    _parsed_fields: List[ParsedAddFieldDefinition] = field(
+        init=False, repr=False, default_factory=list
+    )
+
+    def __post_init__(self) -> None:
+        self._filter_interpolator = InterpolatedBoolean(condition=self.condition, parameters={})
+
+        for add_field in self.fields:
+            if len(add_field.path) < 1:
+                raise ValueError(
+                    f"Expected a non-zero-length path for the AddFields transformation {add_field}"
+                )
+
+            if not isinstance(add_field.value, InterpolatedString):
+                if not isinstance(add_field.value, str):
+                    raise ValueError(
+                        f"Expected a string value for the AddFields transformation: {add_field}"
+                    )
+                else:
+                    self._parsed_fields.append(
+                        ParsedAddFieldDefinition(
+                            add_field.path,
+                            InterpolatedString.create(add_field.value, parameters={}),
+                            value_type=add_field.value_type,
+                        )
+                    )
+            else:
+                self._parsed_fields.append(
+                    ParsedAddFieldDefinition(
+                        add_field.path,
+                        add_field.value,
+                        value_type=add_field.value_type,
+                    )
+                )
+
+    def transform(
+        self,
+        config: MutableMapping[str, Any],
+    ) -> None:
+        """
+        Transforms a config by adding fields based on the provided field definitions.
+
+        :param config: The user-provided configuration to be transformed
+        """
+        for parsed_field in self._parsed_fields:
+            valid_types = (parsed_field.value_type,) if parsed_field.value_type else None
+            value = parsed_field.value.eval(config, valid_types=valid_types)
+            is_empty_condition = not self.condition
+            if is_empty_condition or self._filter_interpolator.eval(
+                config, value=value, path=parsed_field.path
+            ):
+                dpath.new(config, parsed_field.path, value)

airbyte_cdk/sources/declarative/transformations/config_transformations/remove_fields.py

@@ -0,0 +1,72 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+import logging
+from dataclasses import dataclass
+from typing import Any, List, MutableMapping
+
+import dpath
+import dpath.exceptions
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
+from airbyte_cdk.sources.declarative.transformations.config_transformations.config_transformation import (
+    ConfigTransformation,
+)
+from airbyte_cdk.sources.types import FieldPointer
+
+logger = logging.getLogger("airbyte")
+
+
+@dataclass
+class RemoveFields(ConfigTransformation):
+    """
+    A transformation which removes fields from a config. The fields removed are designated using FieldPointers.
+    During transformation, if a field or any of its parents does not exist in the config, no error is thrown.
+
+    If an input field pointer references an item in a list (e.g: ["k", 0] in the object {"k": ["a", "b", "c"]}) then
+    the object at that index is set to None rather than being entirely removed from the list.
+
+    It's possible to remove objects nested in lists e.g: removing [".", 0, "k"] from {".": [{"k": "V"}]} results in {".": [{}]}
+
+    Usage syntax:
+
+    ```yaml
+        config_transformations:
+          - type: RemoveFields
+            field_pointers:
+              - ["path", "to", "field1"]
+              - ["path2"]
+            condition: "{{ config.some_flag }}" # Optional condition
+    ```
+
+    Attributes:
+        field_pointers (List[FieldPointer]): pointers to the fields that should be removed
+        condition (str): Optional condition that determines if the fields should be removed
+    """
+
+    field_pointers: List[FieldPointer]
+    condition: str = ""
+
+    def __post_init__(self) -> None:
+        self._filter_interpolator = InterpolatedBoolean(condition=self.condition, parameters={})
+
+    def transform(
+        self,
+        config: MutableMapping[str, Any],
+    ) -> None:
+        """
+        Transforms a config by removing fields based on the provided field pointers.
+
+        :param config: The user-provided configuration to be transformed
+        """
+        if self.condition and not self._filter_interpolator.eval(config):
+            return
+
+        for pointer in self.field_pointers:
+            try:
+                dpath.delete(config, pointer)
+            except dpath.exceptions.PathNotFound:
+                pass
+            except Exception as e:
+                logger.error(f"Error removing field {pointer}: {e}")
+                raise e

unit_tests/sources/declarative/transformations/config_transformations/test_add_fields.py

@@ -0,0 +1,97 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+import pytest
+
+from airbyte_cdk.sources.declarative.transformations.config_transformations.add_fields import (
+    AddedFieldDefinition,
+    AddFields,
+)
+
+
+def test_given_valid_static_value_fields_added():
+    transformation = AddFields(
+        fields=[
+            AddedFieldDefinition(path=["new_field"], value="static_value"),
+            AddedFieldDefinition(path=["another_field"], value="another_value"),
+        ]
+    )
+    config = {}
+
+    transformation.transform(config)
+
+    assert config == {
+        "new_field": "static_value",
+        "another_field": "another_value",
+    }
+
+
+def test_given_valid_nested_fields_static_value_added():
+    transformation = AddFields(
+        fields=[
+            AddedFieldDefinition(path=["parent", "child", "grandchild"], value="nested_value"),
+        ]
+    )
+    config = {}
+
+    transformation.transform(config)
+
+    assert config == {"parent": {"child": {"grandchild": "nested_value"}}}
+
+
+def test_given_valid_interpolated_input_field_added():
+    transformation = AddFields(
+        fields=[
+            AddedFieldDefinition(path=["derived_field"], value="{{ config.original_field }}"),
+            AddedFieldDefinition(path=["expression_result"], value="{{ 2 * 3 }}"),
+        ]
+    )
+    config = {"original_field": "original_value"}
+
+    transformation.transform(config)
+
+    assert config == {
+        "original_field": "original_value",
+        "derived_field": "original_value",
+        "expression_result": 6,
+    }
+
+
+def test_given_invalid_field_raises_exception():
+    with pytest.raises(ValueError):
+        AddFields(fields=[AddedFieldDefinition(path=[], value="value")])
+
+    with pytest.raises(ValueError):
+        AddFields(fields=[AddedFieldDefinition(path=["valid_path"], value=123)])
+
+
+def test_given_field_already_exists_value_is_overwritten():
+    transformation = AddFields(
+        fields=[
+            AddedFieldDefinition(path=["existing_field"], value="new_value"),
+        ]
+    )
+    config = {"existing_field": "existing_value"}
+
+    transformation.transform(config)
+
+    assert config["existing_field"] == "new_value"
+
+
+def test_with_condition_only_adds_fields_when_condition_is_met():
+    transformation = AddFields(
+        fields=[
+            AddedFieldDefinition(path=["conditional_field"], value="added_value"),
+        ],
+        condition="{{ config.flag == true }}",
+    )
+
+    config_true = {"flag": True}
+    transformation.transform(config_true)
+
+    config_false = {"flag": False}
+    transformation.transform(config_false)
+
+    assert "conditional_field" in config_true
+    assert "conditional_field" not in config_false