mypy fix

Author: darynaishchenko

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -793,6 +793,7 @@ def create_dpath_flatten_fields(
         model_field_path: List[Union[InterpolatedString, str]] = [x for x in model.field_path]
         key_transformation = (
             KeyTransformation(
+                config=config,
                 prefix=model.key_transformation.prefix,
                 suffix=model.key_transformation.suffix,
                 parameters=model.parameters or {},

airbyte_cdk/sources/declarative/transformations/dpath_flatten_fields.py

@@ -10,13 +10,16 @@
 
 @dataclass
 class KeyTransformation:
+    config: Config
     parameters: InitVar[Mapping[str, Any]]
     prefix: Union[InterpolatedString, str, None] = None
     suffix: 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)
+        if self.prefix is not None:
+            self.prefix = InterpolatedString.create(self.prefix, parameters=parameters).eval(self.config)
+        if self.suffix is not None:
+            self.suffix = InterpolatedString.create(self.suffix, parameters=parameters).eval(self.config)
 
 
 @dataclass
@@ -52,12 +55,10 @@ 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:
-                if prefix := self.key_transformation.prefix.eval(config=self.config):
-                    extracted = {f"{prefix}{key}": value for key, value in extracted.items()}
+                extracted = {f"{self.key_transformation.prefix}{key}": value for key, value in extracted.items()}
 
             if self.key_transformation.suffix:
-                if suffix := self.key_transformation.suffix.eval(config=self.config):
-                    extracted = {f"{key}{suffix}": value for key, value in extracted.items()}
+                extracted = {f"{key}{self.key_transformation.suffix}": value for key, value in extracted.items()}
 
         return extracted
 

unit_tests/sources/declarative/transformations/test_dpath_flatten_fields.py

@@ -188,7 +188,7 @@ def test_dpath_flatten_lists(
     expected_record,
 ):
     if key_transformation:
-        key_transformation = KeyTransformation({}, *key_transformation)
+        key_transformation = KeyTransformation(config, {}, *key_transformation)
 
     flattener = DpathFlattenFields(
         field_path=field_path,