update remap to handle interpolated keys/values

Author: pnilan

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

@@ -113,8 +113,7 @@ def transform(
         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(
+            if not self.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/remap_field.py

@@ -2,9 +2,11 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, List, Mapping, MutableMapping, Union
 
+from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
+from airbyte_cdk.sources.declarative.interpolation.interpolated_mapping import InterpolatedMapping
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.transformations.config_transformations.config_transformation import (
     ConfigTransformation,
@@ -19,6 +21,7 @@ class ConfigRemapField(ConfigTransformation):
 
     map: Mapping[str, Any]
     field_path: List[Union[InterpolatedString, str]]
+    config: Mapping[str, Any] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
         if not self.field_path:
@@ -31,6 +34,7 @@ def __post_init__(self) -> None:
                 self._field_path[path_index] = InterpolatedString.create(
                     self.field_path[path_index], parameters={}
                 )
+        self._map = InterpolatedMapping(self.map, parameters={}).eval(config=self.config)
 
     def transform(
         self,
@@ -51,10 +55,10 @@ def transform(
                 return
             current = current[component]
 
-            if not isinstance(current, Mapping):
+            if not isinstance(current, MutableMapping):
                 return
 
         field_name = path_components[-1]
 
-        if field_name in current and current[field_name] in self.map:
-            current[field_name] = self.map[current[field_name]]
+        if field_name in current and current[field_name] in self._map:
+            current[field_name] = self._map[current[field_name]]

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

@@ -3,25 +3,26 @@
 
 import pytest
 
-from airbyte_cdk.sources.declarative.transformations.config_transformations.remap_field import (
+from airbyte_cdk.sources.declarative.transformations.config_transformations import (
     ConfigRemapField as RemapField,
 )
 
 
 class TestRemapField(TestCase):
     def test_given_valid_inputs_when_transform_then_field_is_remapped(self):
-        remap_transform = RemapField(
-            field_path=["authorization", "auth_type"],
-            map={"client_credentials": "oauth2", "api_key": "key_auth"},
-        )
-
         config = {
             "authorization": {
                 "auth_type": "client_credentials",
                 "client_id": "12345",
                 "client_secret": "secret",
             }
         }
+        remap_transform = RemapField(
+            field_path=["authorization", "auth_type"],
+            map={"client_credentials": "oauth2", "api_key": "key_auth"},
+            config=config
+        )
+
         original_config = deepcopy(config)
 
         remap_transform.transform(config)
@@ -110,3 +111,18 @@ def test_multiple_transformations_applied_in_sequence(self):
 
         assert config["auth"]["type"] == "oauth2"
         assert config["environment"] == "development"
+
+    def test_amazon_seller_partner_marketplace_remap_with_interpolated_mapping(self):
+
+        mapping = {
+            "endpoint": {
+                "ES": "{{ 'https://sellingpartnerapi' if config.environment == 'production' else 'https://sandbox.sellingpartnerapi' }}-eu.amazon.com",
+            }
+        }
+        sandbox_config = {"environment": "sandbox", "marketplace": "ES"}
+        production_config = {"environment": "production", "marketplace": "ES"}
+        RemapField(field_path=["marketplace"], map=mapping["endpoint"], config=sandbox_config).transform(sandbox_config)
+        RemapField(field_path=["marketplace"], map=mapping["endpoint"], config=production_config).transform(production_config)
+
+        assert sandbox_config["marketplace"] == "https://sandbox.sellingpartnerapi-eu.amazon.com"
+        assert production_config["marketplace"] == "https://sellingpartnerapi-eu.amazon.com"