feat: add condition to component mapping definition (#667)

Author: pnilan

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -4173,6 +4173,18 @@ definitions:
         description: Determines whether to create a new path if it doesn't exist (true) or only update existing paths (false). When set to true, the resolver will create new paths in the stream template if they don't exist. When false (default), it will only update existing paths.
         type: boolean
         default: false
+      condition:
+        title: Condition
+        description: A condition that must be met for the mapping to be applied. This property is only supported for `ConfigComponentsResolver`.
+        type: string
+        interpolation_context:
+          - config
+          - stream_template_config
+          - components_values
+          - stream_slice
+        examples:
+          - "{{ components_values.get('cursor_field', None) }}"
+          - "{{ '_incremental' in components_values.get('stream_name', '') }}"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -1492,6 +1494,15 @@ class ComponentMappingDefinition(BaseModel):
         description="Determines whether to create a new path if it doesn't exist (true) or only update existing paths (false). When set to true, the resolver will create new paths in the stream template if they don't exist. When false (default), it will only update existing paths.",
         title="Create or Update",
     )
+    condition: Optional[str] = Field(
+        None,
+        description="A condition that must be met for the mapping to be applied. This property is only supported for `ConfigComponentsResolver`.",
+        examples=[
+            "{{ components_values.get('cursor_field', None) }}",
+            "{{ '_incremental' in components_values.get('stream_name', '') }}",
+        ],
+        title="Condition",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3800,6 +3800,7 @@ def create_components_mapping_definition(
             value=interpolated_value,
             value_type=ModelToComponentFactory._json_schema_type_name_to_type(model.value_type),
             create_or_update=model.create_or_update,
+            condition=model.condition,
             parameters=model.parameters or {},
         )
 
@@ -3818,16 +3819,19 @@ def create_http_components_resolver(
             transformations=[],
         )
 
-        components_mapping = [
-            self._create_component_from_model(
-                model=components_mapping_definition_model,
-                value_type=ModelToComponentFactory._json_schema_type_name_to_type(
-                    components_mapping_definition_model.value_type
-                ),
-                config=config,
+        components_mapping = []
+        for component_mapping_definition_model in model.components_mapping:
+            if component_mapping_definition_model.condition:
+                raise ValueError("`condition` is only supported for     `ConfigComponentsResolver`")
+            components_mapping.append(
+                self._create_component_from_model(
+                    model=component_mapping_definition_model,
+                    value_type=ModelToComponentFactory._json_schema_type_name_to_type(
+                        component_mapping_definition_model.value_type
+                    ),
+                    config=config,
+                )
             )
-            for components_mapping_definition_model in model.components_mapping
-        ]
 
         return HttpComponentsResolver(
             retriever=retriever,
@@ -3851,7 +3855,9 @@ def create_stream_config(
         )
 
     def create_config_components_resolver(
-        self, model: ConfigComponentsResolverModel, config: Config
+        self,
+        model: ConfigComponentsResolverModel,
+        config: Config,
     ) -> Any:
         model_stream_configs = (
             model.stream_config if isinstance(model.stream_config, list) else [model.stream_config]
@@ -3871,6 +3877,7 @@ def create_config_components_resolver(
                     components_mapping_definition_model.value_type
                 ),
                 config=config,
+                parameters=model.parameters,
             )
             for components_mapping_definition_model in model.components_mapping
         ]
@@ -3888,16 +3895,20 @@ def create_parametrized_components_resolver(
         stream_parameters = StreamParametersDefinition(
             list_of_parameters_for_stream=model.stream_parameters.list_of_parameters_for_stream
         )
-        components_mapping = [
-            self._create_component_from_model(
-                model=components_mapping_definition_model,
-                value_type=ModelToComponentFactory._json_schema_type_name_to_type(
-                    components_mapping_definition_model.value_type
-                ),
-                config=config,
+
+        components_mapping = []
+        for components_mapping_definition_model in model.components_mapping:
+            if components_mapping_definition_model.condition:
+                raise ValueError("`condition` is only supported for `ConfigComponentsResolver`")
+            components_mapping.append(
+                self._create_component_from_model(
+                    model=components_mapping_definition_model,
+                    value_type=ModelToComponentFactory._json_schema_type_name_to_type(
+                        components_mapping_definition_model.value_type
+                    ),
+                    config=config,
+                )
             )
-            for components_mapping_definition_model in model.components_mapping
-        ]
         return ParametrizedComponentsResolver(
             stream_parameters=stream_parameters,
             config=config,

airbyte_cdk/sources/declarative/resolvers/components_resolver.py

@@ -9,6 +9,7 @@
 from typing_extensions import deprecated
 
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
+from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.source import ExperimentalClassWarning
 
 
@@ -22,6 +23,7 @@ class ComponentMappingDefinition:
     value: Union["InterpolatedString", str]
     value_type: Optional[Type[Any]]
     parameters: InitVar[Mapping[str, Any]]
+    condition: Optional[str] = None
     create_or_update: Optional[bool] = False
 
 
@@ -35,6 +37,7 @@ class ResolvedComponentMappingDefinition:
     value: "InterpolatedString"
     value_type: Optional[Type[Any]]
     parameters: InitVar[Mapping[str, Any]]
+    condition: Optional[InterpolatedBoolean] = None
     create_or_update: Optional[bool] = False
 
 

airbyte_cdk/sources/declarative/resolvers/config_components_resolver.py

@@ -14,6 +14,7 @@
 from yaml.scanner import ScannerError
 
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
+from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.resolvers.components_resolver import (
     ComponentMappingDefinition,
     ComponentsResolver,
@@ -70,6 +71,12 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         """
 
         for component_mapping in self.components_mapping:
+            interpolated_condition = (
+                InterpolatedBoolean(condition=component_mapping.condition, parameters=parameters)
+                if component_mapping.condition
+                else None
+            )
+
             if isinstance(component_mapping.value, (str, InterpolatedString)):
                 interpolated_value = (
                     InterpolatedString.create(component_mapping.value, parameters=parameters)
@@ -89,6 +96,7 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
                         value_type=component_mapping.value_type,
                         create_or_update=component_mapping.create_or_update,
                         parameters=parameters,
+                        condition=interpolated_condition,
                     )
                 )
             else:
@@ -155,6 +163,12 @@ def resolve_components(
             kwargs["components_values"] = components_values  # type: ignore[assignment] # component_values will always be of type Mapping[str, Any]
 
             for resolved_component in self._resolved_components:
+                if (
+                    resolved_component.condition is not None
+                    and not resolved_component.condition.eval(self.config, **kwargs)
+                ):
+                    continue
+
                 valid_types = (
                     (resolved_component.value_type,) if resolved_component.value_type else None
                 )

unit_tests/sources/declarative/resolvers/test_config_components_resolver.py

@@ -145,7 +145,7 @@ def to_configured_catalog(
     STREAM_CONFIG
 ]
 
-# Manifest with component definition with value that is fails when trying
+# Manifest with component definition with value that fails when trying
 # to parse yaml in _parse_yaml_if_possible but generally contains valid string
 _MANIFEST_WITH_SCANNER_ERROR = deepcopy(_MANIFEST)
 _MANIFEST_WITH_SCANNER_ERROR["dynamic_streams"][0]["components_resolver"][
@@ -173,6 +173,7 @@ def to_configured_catalog(
         (_MANIFEST_WITH_STREAM_CONFIGS_LIST, _CONFIG, None, ["item_1", "item_2", "default_item"]),
         (_MANIFEST_WITH_SCANNER_ERROR, _CONFIG, None, ["item_1", "item_2", "default_item"]),
     ],
+    ids=["no_duplicates", "duplicates", "stream_configs_list", "scanner_error"],
 )
 def test_dynamic_streams_read_with_config_components_resolver(
     manifest, config, expected_exception, expected_stream_names
@@ -223,3 +224,164 @@ def test_dynamic_streams_read_with_config_components_resolver(
         assert len(records) == len(expected_stream_names)
         # Use set comparison to avoid relying on deterministic ordering
         assert set(record.stream for record in records) == set(expected_stream_names)
+
+
+# Manifest with condition that always evaluates to true
+_MANIFEST_WITH_TRUE_CONDITION = deepcopy(_MANIFEST)
+_MANIFEST_WITH_TRUE_CONDITION["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].append(
+    {
+        "type": "ComponentMappingDefinition",
+        "field_path": ["retriever", "requester", "$parameters", "always_included"],
+        "value": "true_condition_value",
+        "create_or_update": True,
+        "condition": "{{ True }}",
+    }
+)
+
+# Manifest with condition that always evaluates to false
+_MANIFEST_WITH_FALSE_CONDITION = deepcopy(_MANIFEST)
+_MANIFEST_WITH_FALSE_CONDITION["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].append(
+    {
+        "type": "ComponentMappingDefinition",
+        "field_path": ["retriever", "requester", "$parameters", "never_included"],
+        "value": "false_condition_value",
+        "create_or_update": True,
+        "condition": "{{ False }}",
+    }
+)
+
+# Manifest with condition using components_values that evaluates to true for some items
+_MANIFEST_WITH_COMPONENTS_VALUES_TRUE_CONDITION = deepcopy(_MANIFEST)
+_MANIFEST_WITH_COMPONENTS_VALUES_TRUE_CONDITION["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].append(
+    {
+        "type": "ComponentMappingDefinition",
+        "field_path": ["retriever", "requester", "$parameters", "conditional_param"],
+        "value": "item_1_special_value",
+        "create_or_update": True,
+        "condition": "{{ components_values['name'] == 'item_1' }}",
+    }
+)
+
+# Manifest with condition using components_values that evaluates to false for all items
+_MANIFEST_WITH_COMPONENTS_VALUES_FALSE_CONDITION = deepcopy(_MANIFEST)
+_MANIFEST_WITH_COMPONENTS_VALUES_FALSE_CONDITION["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].append(
+    {
+        "type": "ComponentMappingDefinition",
+        "field_path": ["retriever", "requester", "$parameters", "never_matching"],
+        "value": "never_applied_value",
+        "create_or_update": True,
+        "condition": "{{ components_values['name'] == 'non_existent_item' }}",
+    }
+)
+
+# Manifest with multiple conditions - some true, some false
+_MANIFEST_WITH_MIXED_CONDITIONS = deepcopy(_MANIFEST)
+_MANIFEST_WITH_MIXED_CONDITIONS["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].extend(
+    [
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "always_true"],
+            "value": "always_applied",
+            "create_or_update": True,
+            "condition": "{{ True }}",
+        },
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "always_false"],
+            "value": "never_applied",
+            "create_or_update": True,
+            "condition": "{{ False }}",
+        },
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "item_specific"],
+            "value": "applied_to_item_2",
+            "create_or_update": True,
+            "condition": "{{ components_values['id'] == 2 }}",
+        },
+    ]
+)
+
+
+@pytest.mark.parametrize(
+    "manifest, config, expected_conditional_params",
+    [
+        (
+            _MANIFEST_WITH_TRUE_CONDITION,
+            _CONFIG,
+            {
+                "item_1": {"always_included": "true_condition_value", "item_id": 1},
+                "item_2": {"always_included": "true_condition_value", "item_id": 2},
+                "default_item": {"always_included": "true_condition_value", "item_id": 4},
+            },
+        ),
+        (
+            _MANIFEST_WITH_FALSE_CONDITION,
+            _CONFIG,
+            {
+                "item_1": {"item_id": 1},  # never_included should not be present
+                "item_2": {"item_id": 2},
+                "default_item": {"item_id": 4},
+            },
+        ),
+        (
+            _MANIFEST_WITH_COMPONENTS_VALUES_TRUE_CONDITION,
+            _CONFIG,
+            {
+                "item_1": {"conditional_param": "item_1_special_value", "item_id": 1},
+                "item_2": {"item_id": 2},  # condition false for item_2
+                "default_item": {"item_id": 4},  # condition false for default_item
+            },
+        ),
+        (
+            _MANIFEST_WITH_COMPONENTS_VALUES_FALSE_CONDITION,
+            _CONFIG,
+            {
+                "item_1": {"item_id": 1},  # never_matching should not be present
+                "item_2": {"item_id": 2},
+                "default_item": {"item_id": 4},
+            },
+        ),
+        (
+            _MANIFEST_WITH_MIXED_CONDITIONS,
+            _CONFIG,
+            {
+                "item_1": {"always_true": "always_applied", "item_id": 1},
+                "item_2": {
+                    "always_true": "always_applied",
+                    "item_specific": "applied_to_item_2",
+                    "item_id": 2,
+                },
+                "default_item": {"always_true": "always_applied", "item_id": 4},
+            },
+        ),
+    ],
+    ids=[
+        "true_condition",
+        "false_condition",
+        "components_values_true_condition",
+        "components_values_false_condition",
+        "mixed_conditions",
+    ],
+)
+def test_component_mapping_conditions(manifest, config, expected_conditional_params):
+    """Test that ComponentMappingDefinition conditions work correctly for various scenarios."""
+    source = ConcurrentDeclarativeSource(
+        source_config=manifest, config=config, catalog=None, state=None
+    )
+
+    for stream in source.streams(config):
+        if stream.name in expected_conditional_params:
+            assert (
+                stream.retriever.requester._parameters == expected_conditional_params[stream.name]
+            )