feat(low-code): add regex and copy options to KeysReplace

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3210,7 +3210,30 @@ definitions:
         additionalProperties: true
   KeysReplace:
     title: Keys Replace
-    description: A transformation that replaces symbols in keys.
+    description: >-
+      A transformation that replaces symbols in keys. By default, every occurrence of `old` in each key is
+      replaced with `new` (literal match), the old key is dropped (rename), and the replacement is applied
+      recursively to nested objects of the whole record. `regex`, `keep_original`, `only_if_missing` and
+      `field_path` adjust this behavior. Keys are evaluated against a snapshot of the object, so keys written
+      by the transformation are not processed again. It can also be used in `DynamicSchemaLoader.schema_transformations`,
+      where it is applied to the map of schema properties.
+    examples:
+      - type: KeysReplace
+        old: " "
+        new: "_"
+      - type: KeysReplace
+        old: "hs_v2_date_entered_"
+        new: "hs_date_entered_"
+        keep_original: true
+        only_if_missing: true
+        field_path: ["properties"]
+      - type: KeysReplace
+        old: "hs_v2_date_entered_(.*?)(?:_date)?$"
+        new: "hs_lifecyclestage_\\1_date"
+        regex: true
+        keep_original: true
+        only_if_missing: true
+        field_path: ["properties"]
     type: object
     required:
       - type
@@ -3246,6 +3269,39 @@ definitions:
           - config
           - record
           - stream_slice
+      regex:
+        type: boolean
+        title: Regex
+        description: >-
+          If true, `old` is treated as a regular expression pattern and `new` as a `re.sub` replacement string,
+          which supports backreferences such as `\1` or `\g<1>`. Invalid patterns fail when the component is created.
+        default: false
+      keep_original:
+        type: boolean
+        title: Keep Original
+        description: If true, the value is copied to the new key and the original key is kept, instead of renaming it.
+        default: false
+      only_if_missing:
+        type: boolean
+        title: Only If Missing
+        description: >-
+          If true, the new key is only written when it is absent from the object or its value is null. When the key
+          is not written and `keep_original` is false, the original key is kept unchanged.
+        default: false
+      field_path:
+        type: array
+        title: Field Path
+        description: >-
+          Path to the object whose keys are replaced. Defaults to the whole record. A `*` segment applies the
+          replacement to every matching object. If the path does not exist or does not point to an object,
+          the transformation does nothing.
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["properties"]
+          - ["data", "*", "attributes"]
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -842,6 +842,27 @@ class KeysReplace(BaseModel):
         ],
         title="New value",
     )
+    regex: Optional[bool] = Field(
+        False,
+        description="If true, `old` is treated as a regular expression pattern and `new` as a `re.sub` replacement string, which supports backreferences such as `\\1` or `\\g<1>`. Invalid patterns fail when the component is created.",
+        title="Regex",
+    )
+    keep_original: Optional[bool] = Field(
+        False,
+        description="If true, the value is copied to the new key and the original key is kept, instead of renaming it.",
+        title="Keep Original",
+    )
+    only_if_missing: Optional[bool] = Field(
+        False,
+        description="If true, the new key is only written when it is absent from the object or its value is null. When the key is not written and `keep_original` is false, the original key is kept unchanged.",
+        title="Only If Missing",
+    )
+    field_path: Optional[List[str]] = Field(
+        None,
+        description="Path to the object whose keys are replaced. Defaults to the whole record. A `*` segment applies the replacement to every matching object. If the path does not exist or does not point to an object, the transformation does nothing.",
+        examples=[["properties"], ["data", "*", "attributes"]],
+        title="Field Path",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1103,7 +1103,13 @@ def create_keys_replace_transformation(
         self, model: KeysReplaceModel, config: Config, **kwargs: Any
     ) -> KeysReplaceTransformation:
         return KeysReplaceTransformation(
-            old=model.old, new=model.new, parameters=model.parameters or {}
+            old=model.old,
+            new=model.new,
+            regex=bool(model.regex),
+            keep_original=bool(model.keep_original),
+            only_if_missing=bool(model.only_if_missing),
+            field_path=list(model.field_path) if model.field_path else None,
+            parameters=model.parameters or {},
         )
 
     def create_flatten_fields(

airbyte_cdk/sources/declarative/transformations/keys_replace_transformation.py

@@ -2,14 +2,22 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 #
 
+import copy
+import re
 from dataclasses import InitVar, dataclass
-from typing import Any, Dict, Mapping, Optional
+from typing import Any, Dict, List, Mapping, Optional, Pattern, Union
+
+import dpath
 
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.transformations import RecordTransformation
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
 
 
+def _is_interpolated(value: str) -> bool:
+    return "{{" in value or "{%" in value
+
+
 @dataclass
 class KeysReplaceTransformation(RecordTransformation):
     """
@@ -22,15 +30,64 @@ class KeysReplaceTransformation(RecordTransformation):
     Result:
     from: {"created time": ..., "customer id": ..., "user id": ...}
     to: {"created_time": ..., "customer_id": ..., "user_id": ...}
+
+    By default, `old` is matched literally, matching keys are renamed and the replacement is applied
+    recursively to nested objects of the whole record. Optional flags change this behavior:
+    - regex: `old` is a regular expression and `new` a `re.sub` replacement (backreferences supported).
+    - keep_original: the value is copied to the new key and the original key is kept.
+    - only_if_missing: the new key is only written when it is absent or null.
+    - field_path: the replacement is scoped to the object at this path (`*` matches every element).
+    Keys are evaluated against a snapshot of each object, so keys written by the transformation are not
+    processed again.
     """
 
     old: str
     new: str
     parameters: InitVar[Mapping[str, Any]]
+    regex: bool = False
+    keep_original: bool = False
+    only_if_missing: bool = False
+    field_path: Optional[List[Union[InterpolatedString, str]]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._old = InterpolatedString.create(self.old, parameters=parameters)
         self._new = InterpolatedString.create(self.new, parameters=parameters)
+        self._field_path = [
+            InterpolatedString.create(path, parameters=parameters)
+            for path in (self.field_path or [])
+        ]
+        self._static_pattern: Optional[Pattern[str]] = None
+        if self.regex and not _is_interpolated(self.old):
+            self._static_pattern = self._compile_pattern(self.old)
+            if not _is_interpolated(self.new):
+                self._validate_replacement(self._static_pattern, self.new)
+
+    @staticmethod
+    def _compile_pattern(pattern: str) -> Pattern[str]:
+        try:
+            return re.compile(pattern)
+        except re.error as exception:
+            raise ValueError(
+                f"KeysReplace `old` is not a valid regular expression: {pattern!r} ({exception})."
+            ) from exception
+
+    @staticmethod
+    def _validate_replacement(pattern: Pattern[str], replacement: str) -> None:
+        try:
+            pattern.sub(replacement, "")
+        except (re.error, IndexError) as exception:
+            raise ValueError(
+                f"KeysReplace `new` is not a valid replacement for pattern {pattern.pattern!r}: {replacement!r} ({exception})."
+            ) from exception
+
+    def _eval_regex_part(
+        self, raw: str, interpolated: InterpolatedString, config: Config, **kwargs: Any
+    ) -> str:
+        # Plain strings are used as-is: Jinja evaluation runs `ast.literal_eval` on the output,
+        # which would turn a pattern like `(1)` into `1`.
+        if not _is_interpolated(raw):
+            return raw
+        return str(interpolated.eval(config, valid_types=(str,), **kwargs))
 
     def transform(
         self,
@@ -43,19 +100,67 @@ def transform(
             config = {}
 
         kwargs = {"record": record, "stream_state": stream_state, "stream_slice": stream_slice}
-        old_key = str(self._old.eval(config, **kwargs))
-        new_key = str(self._new.eval(config, **kwargs))
+        if self.regex:
+            pattern = self._static_pattern or self._compile_pattern(
+                self._eval_regex_part(self.old, self._old, config, **kwargs)
+            )
+            replacement = self._eval_regex_part(self.new, self._new, config, **kwargs)
+            if self._static_pattern is None or _is_interpolated(self.new):
+                self._validate_replacement(pattern, replacement)
+
+            def replace_key(key: str) -> str:
+                return pattern.sub(replacement, key)
+        else:
+            old_key = str(self._old.eval(config, **kwargs))
+            new_key = str(self._new.eval(config, **kwargs))
+
+            def replace_key(key: str) -> str:
+                return key.replace(old_key, new_key)
 
         def _transform(data: Dict[str, Any]) -> Dict[str, Any]:
+            transformed = {
+                key: _transform(value) if isinstance(value, dict) else value
+                for key, value in data.items()
+            }
+
+            if self.keep_original:
+                result = dict(transformed)
+                for key, value in transformed.items():
+                    updated_key = replace_key(key)
+                    if updated_key == key:
+                        continue
+                    if self.only_if_missing and result.get(updated_key) is not None:
+                        continue
+                    result[updated_key] = (
+                        copy.deepcopy(value) if isinstance(value, (dict, list)) else value
+                    )
+                return result
+
             result = {}
-            for key, value in data.items():
-                updated_key = key.replace(old_key, new_key)
-                if isinstance(value, dict):
-                    result[updated_key] = _transform(value)
-                else:
-                    result[updated_key] = value
+            for key, value in transformed.items():
+                updated_key = replace_key(key)
+                if self.only_if_missing:
+                    if updated_key != key and (
+                        result.get(updated_key) is not None
+                        or transformed.get(updated_key) is not None
+                    ):
+                        updated_key = key
+                    if value is None and result.get(updated_key) is not None:
+                        continue
+                result[updated_key] = value
             return result
 
-        transformed_record = _transform(record)
-        record.clear()
-        record.update(transformed_record)
+        for target in self._get_targets(record, config):
+            transformed_target = _transform(target)
+            target.clear()
+            target.update(transformed_target)
+
+    def _get_targets(self, record: Dict[str, Any], config: Config) -> List[Dict[str, Any]]:
+        if not self._field_path:
+            return [record]
+        path = [str(path.eval(config)) for path in self._field_path]
+        if "*" in path:
+            matches = dpath.values(record, path)
+        else:
+            matches = [dpath.get(record, path, default=None)]
+        return [match for match in matches if isinstance(match, dict)]

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -144,6 +144,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     JsonlDecoder as JsonlDecoderModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    KeysReplace as KeysReplaceModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     OffsetIncrement as OffsetIncrementModel,
 )
@@ -10096,3 +10099,77 @@ def test_create_file_uploader_with_a_combined_download_target_extractor():
     )
 
     assert isinstance(file_uploader.download_target_extractor, CombinedExtractor)
+
+
+def test_create_keys_replace_transformation_defaults():
+    transformation = factory.create_component(
+        model_type=KeysReplaceModel,
+        component_definition={"type": "KeysReplace", "old": " ", "new": "_"},
+        config=input_config,
+    )
+
+    assert isinstance(transformation, KeysReplaceTransformation)
+    assert transformation.regex is False
+    assert transformation.keep_original is False
+    assert transformation.only_if_missing is False
+    assert transformation.field_path is None
+
+
+def test_create_keys_replace_transformation_with_options():
+    content = r"""
+    type: KeysReplace
+    old: "hs_v2_date_entered_(.*?)(?:_date)?$"
+    new: "hs_lifecyclestage_\\1_date"
+    regex: true
+    keep_original: true
+    only_if_missing: true
+    field_path: ["properties"]
+    """
+    definition = YamlDeclarativeSource._parse(content)
+
+    transformation = factory.create_component(
+        model_type=KeysReplaceModel, component_definition=definition, config=input_config
+    )
+
+    assert isinstance(transformation, KeysReplaceTransformation)
+    assert transformation.new == r"hs_lifecyclestage_\1_date"
+    assert transformation.regex is True
+    assert transformation.keep_original is True
+    assert transformation.only_if_missing is True
+    assert transformation.field_path == ["properties"]
+
+    record = {
+        "id": 1,
+        "properties": {
+            "hs_v2_date_entered_customer": "2024-01-01",
+            "hs_v2_date_entered_lead_date": "2024-02-01",
+            "hs_v2_date_entered_opportunity": "2024-03-01",
+            "hs_lifecyclestage_opportunity_date": "2023-12-31",
+        },
+    }
+    transformation.transform(record, config=input_config)
+    assert record == {
+        "id": 1,
+        "properties": {
+            "hs_v2_date_entered_customer": "2024-01-01",
+            "hs_v2_date_entered_lead_date": "2024-02-01",
+            "hs_v2_date_entered_opportunity": "2024-03-01",
+            "hs_lifecyclestage_customer_date": "2024-01-01",
+            "hs_lifecyclestage_lead_date": "2024-02-01",
+            "hs_lifecyclestage_opportunity_date": "2023-12-31",
+        },
+    }
+
+
+def test_create_keys_replace_transformation_with_invalid_regex_raises():
+    with pytest.raises(ValueError, match="KeysReplace `old` is not a valid regular expression"):
+        factory.create_component(
+            model_type=KeysReplaceModel,
+            component_definition={
+                "type": "KeysReplace",
+                "old": "(unclosed",
+                "new": "x",
+                "regex": True,
+            },
+            config=input_config,
+        )