feat(low-code): added key_transformation to DpathFlattenFields

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2335,6 +2335,12 @@ definitions:
         title: Replace Origin Record
         description: Whether to replace the origin record or not. Default is False.
         type: boolean
+      key_transformation:
+        title: Key transformation
+        description: Transformation for object keys. If not provided, original key will be used.
+        type: string
+        examples:
+          - "flattened_{{ key }}"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -897,6 +897,14 @@ class DpathFlattenFields(BaseModel):
         description="Whether to replace the origin record or not. Default is False.",
         title="Replace Origin Record",
     )
+    key_transformation: Optional[str] = Field(
+        None,
+        description="Transformation for object keys. If not provided, original key will be used.",
+        examples=[
+            "flattened_{{ key }}",
+        ],
+        title="Key transformation",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -797,6 +797,7 @@ def create_dpath_flatten_fields(
             if model.delete_origin_value is not None
             else False,
             replace_record=model.replace_record if model.replace_record is not None else False,
+            key_transformation=model.key_transformation,
             parameters=model.parameters or {},
         )
 

airbyte_cdk/sources/declarative/transformations/dpath_flatten_fields.py

@@ -16,6 +16,7 @@ class DpathFlattenFields(RecordTransformation):
     field_path: List[Union[InterpolatedString, str]] path to the field to flatten.
     delete_origin_value: bool = False whether to delete origin field or keep it. Default is False.
     replace_record: bool = False whether to replace origin record or not. Default is False.
+    key_transformation: string = None how to transform extracted object keys
 
     """
 
@@ -24,6 +25,7 @@ class DpathFlattenFields(RecordTransformation):
     parameters: InitVar[Mapping[str, Any]]
     delete_origin_value: bool = False
     replace_record: bool = False
+    key_transformation: Union[InterpolatedString, str] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._field_path = [
@@ -34,6 +36,7 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
                 self._field_path[path_index] = InterpolatedString.create(
                     self.field_path[path_index], parameters=parameters
                 )
+        self.parameters = parameters
 
     def transform(
         self,
@@ -50,6 +53,14 @@ def transform(
             extracted = dpath.get(record, path, default=[])
 
         if isinstance(extracted, dict):
+
+            if self.key_transformation:
+                updated_extracted = {}
+                for key, value in extracted.items():
+                    updated_key = InterpolatedString.create(self.key_transformation, parameters=self.parameters).eval(key=key, config=self.config)
+                    updated_extracted[updated_key] = value
+                extracted = updated_extracted
+
             if self.replace_record and extracted:
                 dpath.delete(record, "**")
                 record.update(extracted)

unit_tests/sources/declarative/transformations/test_dpath_flatten_fields.py

@@ -7,7 +7,7 @@
 _REPLACE_WITH_VALUE = True
 _DO_NOT_DELETE_ORIGIN_VALUE = False
 _DO_NOT_REPLACE_WITH_VALUE = False
-
+_NO_KEY_TRANSFORMATIONS = None
 
 @pytest.mark.parametrize(
     [
@@ -16,6 +16,7 @@
         "field_path",
         "delete_origin_value",
         "replace_record",
+        "key_transformation",
         "expected_record",
     ],
     [
@@ -25,6 +26,7 @@
             ["field2"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE},
             id="flatten by dpath, don't delete origin value",
         ),
@@ -34,6 +36,7 @@
             ["field2"],
             _DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field3": _ANY_VALUE},
             id="flatten by dpath, delete origin value",
         ),
@@ -46,6 +49,7 @@
             ["field2", "*", "field4"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {
                 "field1": _ANY_VALUE,
                 "field2": {"field3": {"field4": {"field5": _ANY_VALUE}}},
@@ -62,6 +66,7 @@
             ["field2", "*", "field4"],
             _DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field2": {"field3": {}}, "field5": _ANY_VALUE},
             id="flatten by dpath with *, delete origin value",
         ),
@@ -71,6 +76,7 @@
             ["{{ config['field_path'] }}"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE},
             id="flatten by dpath from config, don't delete origin value",
         ),
@@ -80,6 +86,7 @@
             ["non-existing-field"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
             id="flatten by non-existing dpath, don't delete origin value",
         ),
@@ -89,6 +96,7 @@
             ["*", "non-existing-field"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
             id="flatten by non-existing dpath with *, don't delete origin value",
         ),
@@ -98,6 +106,7 @@
             ["field2"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE},
             id="flatten by dpath, not to update when record has field conflicts, don't delete origin value",
         ),
@@ -107,6 +116,7 @@
             ["field2"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _DO_NOT_REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE},
             id="flatten by dpath, not to update when record has field conflicts, delete origin value",
         ),
@@ -116,6 +126,7 @@
             ["field2"],
             _DO_NOT_DELETE_ORIGIN_VALUE,
             _REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"field3": _ANY_VALUE},
             id="flatten by dpath, replace with value",
         ),
@@ -125,20 +136,52 @@
             ["field2"],
             _DELETE_ORIGIN_VALUE,
             _REPLACE_WITH_VALUE,
+            _NO_KEY_TRANSFORMATIONS,
             {"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,
+            "field2_{{ key }}",
+            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field2_field3": _ANY_VALUE},
+            id="flatten by dpath, not delete origin value, add keys transformation",
+        ),
+        pytest.param(
+            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
+            {},
+            ["field2"],
+            _DELETE_ORIGIN_VALUE,
+            _DO_NOT_REPLACE_WITH_VALUE,
+            "field2_{{ key }}",
+            {"field1": _ANY_VALUE, "field2_field3": _ANY_VALUE},
+            id="flatten by dpath, delete origin value, add keys transformation",
+        ),
+        pytest.param(
+            {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}},
+            {},
+            ["field2"],
+            _DO_NOT_DELETE_ORIGIN_VALUE,
+            _REPLACE_WITH_VALUE,
+            "field2_{{ key }}",
+            {"field2_field3": _ANY_VALUE},
+            id="flatten by dpath, not delete origin value, replace record, add keys transformation",
+        ),
     ],
 )
 def test_dpath_flatten_lists(
-    input_record, config, field_path, delete_origin_value, replace_record, expected_record
+    input_record, config, field_path, delete_origin_value, replace_record, key_transformation, expected_record
 ):
     flattener = DpathFlattenFields(
         field_path=field_path,
         parameters={},
         config=config,
         delete_origin_value=delete_origin_value,
         replace_record=replace_record,
+        key_transformation=key_transformation,
     )
     flattener.transform(input_record)
     assert input_record == expected_record