removed custom option in KeyTransformation

Author: darynaishchenko

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2328,12 +2328,6 @@ definitions:
         type: string
         examples:
           - _flattened
-      custom:
-        title: Custom Transformation for keys
-        description: Custom transformation. Can be used with {{ key }} as a original value for key name. If not provided original keys remain unchanged.
-        type: string
-        examples:
-          - flattened_{{ key }}
   DpathFlattenFields:
     title: Dpath Flatten Fields
     description: A transformation that flatten field values to the to top of the record.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -896,14 +896,6 @@ class KeyTransformation(BaseModel):
         ],
         title="Key Suffix",
     )
-    custom: Optional[Union[str, None]] = Field(
-        None,
-        description="Custom transformation. Can be used with {{ key }} as a original value for key name. If not provided original keys remain unchanged.",
-        examples=[
-            "flattened_{{ key }}",
-        ],
-        title="Custom Transformation for keys",
-    )
 
 
 class DpathFlattenFields(BaseModel):

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -795,7 +795,7 @@ def create_dpath_flatten_fields(
             KeyTransformation(
                 prefix=model.key_transformation.prefix,
                 suffix=model.key_transformation.suffix,
-                custom=model.key_transformation.custom,
+                parameters=model.parameters or {},
             )
             if model.key_transformation is not None
             else None

airbyte_cdk/sources/declarative/transformations/dpath_flatten_fields.py

@@ -8,11 +8,15 @@
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
 
 
-@dataclass(frozen=True)
+@dataclass
 class KeyTransformation:
+    parameters: InitVar[Mapping[str, Any]]
     prefix: Union[InterpolatedString, str, None] = None
     suffix: Union[InterpolatedString, str, None] = None
-    custom: Union[InterpolatedString, str, None] = None
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self.prefix = InterpolatedString.create(self.prefix, parameters=parameters)
+        self.suffix = InterpolatedString.create(self.suffix, parameters=parameters)
 
 
 @dataclass
@@ -32,7 +36,7 @@ class DpathFlattenFields(RecordTransformation):
     parameters: InitVar[Mapping[str, Any]]
     delete_origin_value: bool = False
     replace_record: bool = False
-    key_transformation: Union[KeyTransformation, None] = None
+    key_transformation: Optional[KeyTransformation] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._parameters = parameters
@@ -48,25 +52,13 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
     def _apply_key_transformation(self, extracted: Mapping[str, Any]) -> Mapping[str, Any]:
         if self.key_transformation:
             if self.key_transformation.prefix:
-                prefix = InterpolatedString.create(
-                    self.key_transformation.prefix, parameters=self._parameters
-                ).eval(config=self.config)
-                extracted = {f"{prefix}{key}": value for key, value in extracted.items()}
+                if prefix := self.key_transformation.prefix.eval(config=self.config):
+                    extracted = {f"{prefix}{key}": value for key, value in extracted.items()}
 
             if self.key_transformation.suffix:
-                suffix = InterpolatedString.create(
-                    self.key_transformation.suffix, parameters=self._parameters
-                ).eval(config=self.config)
-                extracted = {f"{key}{suffix}": value for key, value in extracted.items()}
-
-            if self.key_transformation.custom:
-                updated_extracted = {}
-                for key, value in extracted.items():
-                    updated_key = InterpolatedString.create(
-                        self.key_transformation.custom, parameters=self._parameters
-                    ).eval(key=key, config=self.config)
-                    updated_extracted[updated_key] = value
-                extracted = updated_extracted
+                if suffix := self.key_transformation.suffix.eval(config=self.config):
+                    extracted = {f"{key}{suffix}": value for key, value in extracted.items()}
+
         return extracted
 
     def transform(

unit_tests/sources/declarative/transformations/test_dpath_flatten_fields.py

@@ -12,7 +12,6 @@
 _DO_NOT_REPLACE_WITH_VALUE = False
 _NO_KEY_PREFIX = None
 _NO_KEY_SUFFIX = None
-_NO_CUSTOM_TRANSFORMATION = None
 _NO_KEY_TRANSFORMATIONS = None
 
 
@@ -147,43 +146,13 @@
             {"field3": _ANY_VALUE},
             id="flatten by dpath, delete_origin_value do not affect to replace_record",
         ),
-        pytest.param(
-            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
-            {},
-            ["field2"],
-            _DO_NOT_DELETE_ORIGIN_VALUE,
-            _DO_NOT_REPLACE_WITH_VALUE,
-            (_NO_KEY_PREFIX, _NO_KEY_SUFFIX, "field2_{{ key }}"),
-            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field2_field3": _ANY_VALUE},
-            id="flatten by dpath, not delete origin value, add keys custom transformation",
-        ),
-        pytest.param(
-            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
-            {},
-            ["field2"],
-            _DELETE_ORIGIN_VALUE,
-            _DO_NOT_REPLACE_WITH_VALUE,
-            (_NO_KEY_PREFIX, _NO_KEY_SUFFIX, "field2_{{ key }}"),
-            {"field1": _ANY_VALUE, "field2_field3": _ANY_VALUE},
-            id="flatten by dpath, delete origin value, add keys custom transformation",
-        ),
         pytest.param(
             {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
             {},
             ["field2"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _REPLACE_WITH_VALUE,
-            (_NO_KEY_PREFIX, _NO_KEY_SUFFIX, "field2_{{ key }}"),
-            {"field2_field3": _ANY_VALUE},
-            id="flatten by dpath, not delete origin value, replace record, add keys custom transformation",
-        ),
-        pytest.param(
-            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
-            {},
-            ["field2"],
-            _DO_NOT_DELETE_ORIGIN_VALUE,
-            _REPLACE_WITH_VALUE,
-            ("prefix_", _NO_KEY_SUFFIX, _NO_CUSTOM_TRANSFORMATION),
+            ("prefix_", _NO_KEY_SUFFIX),
             {"prefix_field3": _ANY_VALUE},
             id="flatten by dpath, not delete origin value, replace record, add keys prefix",
         ),
@@ -193,7 +162,7 @@
             ["field2"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _REPLACE_WITH_VALUE,
-            (_NO_KEY_PREFIX, "_suffix", _NO_CUSTOM_TRANSFORMATION),
+            (_NO_KEY_PREFIX, "_suffix"),
             {"field3_suffix": _ANY_VALUE},
             id="flatten by dpath, not delete origin value, replace record, add keys suffix",
         ),
@@ -203,20 +172,10 @@
             ["field2"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _REPLACE_WITH_VALUE,
-            ("prefix_", "_suffix", _NO_CUSTOM_TRANSFORMATION),
+            ("prefix_", "_suffix"),
             {"prefix_field3_suffix": _ANY_VALUE},
             id="flatten by dpath, not delete origin value, replace record, add keys prefix and suffix",
         ),
-        pytest.param(
-            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
-            {},
-            ["field2"],
-            _DO_NOT_DELETE_ORIGIN_VALUE,
-            _REPLACE_WITH_VALUE,
-            ("prefix_", "_suffix", "{{ key|upper }}"),
-            {"PREFIX_FIELD3_SUFFIX": _ANY_VALUE},
-            id="flatten by dpath, not delete origin value, replace record, add keys prefix and suffix with custom to upper case",
-        ),
     ],
 )
 def test_dpath_flatten_lists(
@@ -229,7 +188,7 @@ def test_dpath_flatten_lists(
     expected_record,
 ):
     if key_transformation:
-        key_transformation = KeyTransformation(*key_transformation)
+        key_transformation = KeyTransformation({}, *key_transformation)
 
     flattener = DpathFlattenFields(
         field_path=field_path,