From 98ebee83210cee70f1a0b180a33edb5914b7d26a Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 15 Sep 2026 19:23:34 +0300 Subject: [PATCH 1/4] feat: add ForEach transformation for transforming nested collections Connectors that need to apply the same fix-up to every element of a nested collection have no declarative way to write a loop, so they drop to a CustomTransformation. A single Python class forfeits the Connector Builder for the entire connector, and it does so for logic that is otherwise a plain AddFields. ForEach resolves a dpath field_path to a collection inside the record and runs its nested transformations against each element. RecordTransformation.transform mutates in place, so rebinding the nested call's `record` argument to the element mutates that element inside the parent record - no copy and no copy-back, which is what keeps this cheap and safe. source-monday's MondayTransformation (components.py:463-472) collapses entirely into YAML, and the release-asset loop in source-github's ReleasesRecordTransformation (components.py:148-151) becomes an AddFields plus a RemoveFields inside a ForEach. A missing or non-collection field_path is a silent no-op, because a collection that is absent from some records is normal and raising would fail the sync. A non-object element raises a config-flavoured AirbyteTracedException naming the stream and the path, because silently skipping it would hide a manifest mistake behind missing data. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 53 +++ .../models/declarative_component_schema.py | 47 +++ .../parsers/model_to_component_factory.py | 21 ++ .../declarative/transformations/__init__.py | 3 +- .../declarative/transformations/for_each.py | 133 ++++++++ .../test_model_to_component_factory.py | 83 ++++- .../transformations/test_for_each.py | 308 ++++++++++++++++++ 7 files changed, 646 insertions(+), 2 deletions(-) create mode 100644 airbyte_cdk/sources/declarative/transformations/for_each.py create mode 100644 unit_tests/sources/declarative/transformations/test_for_each.py diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index aa3bbbc5e0..4d55e5235f 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -1867,6 +1867,7 @@ definitions: - "$ref": "#/definitions/FlattenFields" - "$ref": "#/definitions/DpathFlattenFields" - "$ref": "#/definitions/KeysReplace" + - "$ref": "#/definitions/ForEach" - "$ref": "#/definitions/CustomTransformation" state_migrations: title: State Migrations @@ -2833,6 +2834,7 @@ definitions: - "$ref": "#/definitions/FlattenFields" - "$ref": "#/definitions/DpathFlattenFields" - "$ref": "#/definitions/KeysReplace" + - "$ref": "#/definitions/ForEach" - "$ref": "#/definitions/CustomTransformation" schema_type_identifier: "$ref": "#/definitions/SchemaTypeIdentifier" @@ -2955,6 +2957,7 @@ definitions: - "$ref": "#/definitions/FlattenFields" - "$ref": "#/definitions/DpathFlattenFields" - "$ref": "#/definitions/KeysReplace" + - "$ref": "#/definitions/ForEach" - "$ref": "#/definitions/CustomTransformation" $parameters: type: object @@ -3003,6 +3006,56 @@ definitions: $parameters: type: object additionalProperties: true + ForEach: + title: For Each + description: A transformation that applies a list of nested transformations to every element of a collection inside the record, instead of applying them to the record as a whole. Use it when an API returns a list of sub-objects that each need the same fix-up, which would otherwise require a custom Python transformation. + type: object + required: + - type + - field_path + - transformations + properties: + type: + type: string + enum: [ForEach] + field_path: + title: Field Path + description: Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. A `*` wildcard segment is supported, in which case every matching value is iterated over. An element that is not an object raises an error, because it cannot be transformed in place. + type: array + items: + type: string + examples: + - ["column_values"] + - ["data", "assets"] + - ["data", "*", "items"] + - ["{{ parameters['collection_field'] }}"] + transformations: + title: Transformations + description: A list of transformations applied to each element of the collection. Inside these transformations, `record` refers to the current element of the collection, while `config`, `stream_state` and `stream_slice` still refer to the enclosing stream. + type: array + items: + anyOf: + - "$ref": "#/definitions/AddFields" + - "$ref": "#/definitions/RemoveFields" + - "$ref": "#/definitions/KeysToLower" + - "$ref": "#/definitions/KeysToSnakeCase" + - "$ref": "#/definitions/FlattenFields" + - "$ref": "#/definitions/DpathFlattenFields" + - "$ref": "#/definitions/KeysReplace" + - "$ref": "#/definitions/ForEach" + - "$ref": "#/definitions/CustomTransformation" + examples: + - - type: AddFields + condition: "{{ record.get('display_value') and not record.get('text') }}" + fields: + - path: ["text"] + value: "{{ record['display_value'] }}" + - - type: RemoveFields + field_pointers: + - ["uploader"] + $parameters: + type: object + additionalProperties: true KeyTransformation: title: Transformation to apply for extracted object keys by Dpath Flatten Fields type: object diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 21a92cc3be..5ad4938070 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -2331,6 +2331,49 @@ class SessionTokenRequestApiKeyAuthenticator(BaseModel): ) +class ForEach(BaseModel): + type: Literal["ForEach"] + field_path: List[str] = Field( + ..., + description="Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. A `*` wildcard segment is supported, in which case every matching value is iterated over. An element that is not an object raises an error, because it cannot be transformed in place.", + examples=[ + ["column_values"], + ["data", "assets"], + ["data", "*", "items"], + ["{{ parameters['collection_field'] }}"], + ], + title="Field Path", + ) + transformations: List[ + Union[ + AddFields, + RemoveFields, + KeysToLower, + KeysToSnakeCase, + FlattenFields, + DpathFlattenFields, + KeysReplace, + ForEach, + CustomTransformation, + ] + ] = Field( + ..., + description="A list of transformations applied to each element of the collection. Inside these transformations, `record` refers to the current element of the collection, while `config`, `stream_state` and `stream_slice` still refer to the enclosing stream.", + examples=[ + [ + { + "type": "AddFields", + "condition": "{{ record.get('display_value') and not record.get('text') }}", + "fields": [{"path": ["text"], "value": "{{ record['display_value'] }}"}], + } + ], + [{"type": "RemoveFields", "field_pointers": [["uploader"]]}], + ], + title="Transformations", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class JsonSchemaPropertySelector(BaseModel): type: Literal["JsonSchemaPropertySelector"] transformations: Optional[ @@ -2343,6 +2386,7 @@ class JsonSchemaPropertySelector(BaseModel): FlattenFields, DpathFlattenFields, KeysReplace, + ForEach, CustomTransformation, ] ] @@ -2842,6 +2886,7 @@ class Config: FlattenFields, DpathFlattenFields, KeysReplace, + ForEach, CustomTransformation, ] ] @@ -3074,6 +3119,7 @@ class DynamicSchemaLoader(BaseModel): FlattenFields, DpathFlattenFields, KeysReplace, + ForEach, CustomTransformation, ] ] @@ -3476,6 +3522,7 @@ class DynamicDeclarativeStream(BaseModel): ComplexFieldType.update_forward_refs() +ForEach.update_forward_refs() GzipDecoder.update_forward_refs() CompositeErrorHandler.update_forward_refs() DeclarativeSource1.update_forward_refs() diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index aa546c73ab..40474c36aa 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -288,6 +288,9 @@ from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( FlattenFields as FlattenFieldsModel, ) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + ForEach as ForEachModel, +) from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( GroupByKeyMergeStrategy as GroupByKeyMergeStrategyModel, ) @@ -624,6 +627,9 @@ from airbyte_cdk.sources.declarative.transformations.flatten_fields import ( FlattenFields, ) +from airbyte_cdk.sources.declarative.transformations.for_each import ( + ForEach, +) from airbyte_cdk.sources.declarative.transformations.keys_replace_transformation import ( KeysReplaceTransformation, ) @@ -803,6 +809,7 @@ def _init_mappings(self) -> None: KeysReplaceModel: self.create_keys_replace_transformation, FlattenFieldsModel: self.create_flatten_fields, DpathFlattenFieldsModel: self.create_dpath_flatten_fields, + ForEachModel: self.create_for_each, IterableDecoderModel: self.create_iterable_decoder, XmlDecoderModel: self.create_xml_decoder, JsonFileSchemaLoaderModel: self.create_json_file_schema_loader, @@ -1066,6 +1073,20 @@ def create_flatten_fields( flatten_lists=model.flatten_lists if model.flatten_lists is not None else True ) + def create_for_each(self, model: ForEachModel, config: Config, **kwargs: Any) -> ForEach: + transformations: List[RecordTransformation] = [] + for transformation_model in model.transformations: + transformations.append( + self._create_component_from_model(model=transformation_model, config=config) + ) + + return ForEach( + config=config, + field_path=[x for x in model.field_path], + transformations=transformations, + parameters=model.parameters or {}, + ) + def create_dpath_flatten_fields( self, model: DpathFlattenFieldsModel, config: Config, **kwargs: Any ) -> DpathFlattenFields: diff --git a/airbyte_cdk/sources/declarative/transformations/__init__.py b/airbyte_cdk/sources/declarative/transformations/__init__.py index e18712a012..3a1a2a309e 100644 --- a/airbyte_cdk/sources/declarative/transformations/__init__.py +++ b/airbyte_cdk/sources/declarative/transformations/__init__.py @@ -12,6 +12,7 @@ # isort: split from .add_fields import AddFields +from .for_each import ForEach from .remove_fields import RemoveFields -__all__ = ["AddFields", "RecordTransformation", "RemoveFields"] +__all__ = ["AddFields", "ForEach", "RecordTransformation", "RemoveFields"] diff --git a/airbyte_cdk/sources/declarative/transformations/for_each.py b/airbyte_cdk/sources/declarative/transformations/for_each.py new file mode 100644 index 0000000000..3ad89fc94c --- /dev/null +++ b/airbyte_cdk/sources/declarative/transformations/for_each.py @@ -0,0 +1,133 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +from dataclasses import InitVar, dataclass +from typing import Any, Dict, List, Mapping, Optional, Union, cast + +import dpath + +from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString +from airbyte_cdk.sources.declarative.transformations import RecordTransformation +from airbyte_cdk.sources.types import Config, StreamSlice, StreamState +from airbyte_cdk.utils.traced_exception import AirbyteTracedException + + +@dataclass +class ForEach(RecordTransformation): + """ + Applies a list of nested transformations to every element of a collection nested inside the record, + instead of applying them to the record as a whole. + + `RecordTransformation.transform` mutates the record in place, so binding the nested transformations to a + collection element mutates that element inside the parent record. Nothing is copied and nothing needs to + be written back. + + Behavior of `field_path` resolution: + * The path entries are interpolated strings, like every other dpath-based component. + * A `*` wildcard segment is supported. When the path contains a `*`, every value matching the glob is + treated as its own collection to iterate over. + * If the path does not resolve, the transformation is a no-op. A collection missing from some records is + expected and must not fail the sync. + * If the resolved value is a list, the nested transformations are applied to each element. + * If the resolved value is an object, it is treated as a collection of one, consistent with how + `DpathExtractor` treats an object as a single record. + * If the resolved value is a scalar or `None`, the transformation is a no-op. + * An empty `field_path` resolves to the record itself, the same way `DpathExtractor` treats an empty + `field_path`. + * If an element of the resolved list is not an object (for example a list of strings), an error is + raised. Such an element cannot be mutated in place, and silently skipping it would hide a manifest + mistake behind missing data. + + Example: + ```yaml + transformations: + - type: ForEach + field_path: ["column_values"] + transformations: + - type: AddFields + condition: "{{ record.get('display_value') and not record.get('text') }}" + fields: + - path: ["text"] + value: "{{ record['display_value'] }}" + ``` + + Attributes: + config (Config): The user-provided configuration as specified by the source's spec + field_path (List[Union[InterpolatedString, str]]): Path to the collection to iterate over + transformations (List[RecordTransformation]): Transformations applied to each element of the collection + """ + + config: Config + field_path: List[Union[InterpolatedString, str]] + transformations: List[RecordTransformation] + parameters: InitVar[Mapping[str, Any]] + + def __post_init__(self, parameters: Mapping[str, Any]) -> None: + self._parameters = parameters + self._field_path: List[InterpolatedString] = [ + InterpolatedString.create(path, parameters=parameters) for path in self.field_path + ] + + def transform( + self, + record: Dict[str, Any], + config: Optional[Config] = None, + stream_state: Optional[StreamState] = None, + stream_slice: Optional[StreamSlice] = None, + ) -> None: + effective_config = config if config is not None else self.config + path = [path.eval(effective_config) for path in self._field_path] + + for collection in self._resolve_collections(record, path): + for element in self._iterate(collection, path): + for transformation in self.transformations: + transformation.transform( + element, + config=effective_config, + stream_state=stream_state, + stream_slice=stream_slice, + ) + + @staticmethod + def _resolve_collections(record: Dict[str, Any], path: List[str]) -> List[Any]: + """ + Resolve `path` to the collections to iterate over. `dpath` returns references to the nested objects, + never copies, which is what makes the in-place mutation work. + """ + if not path: + return [record] + if "*" in path: + return list(dpath.values(record, path)) + resolved = dpath.get(record, path, default=None) + return [] if resolved is None else [resolved] + + def _iterate(self, collection: Any, path: List[str]) -> List[Dict[str, Any]]: + if isinstance(collection, dict): + return [collection] + if not isinstance(collection, list): + # A scalar cannot hold nested records, so there is nothing to transform. + return [] + + for element in collection: + if not isinstance(element, dict): + stream_name = self._parameters.get("name") + stream_context = f" of stream `{stream_name}`" if stream_name else "" + raise AirbyteTracedException( + message=( + f"The ForEach transformation{stream_context} could not be applied because the " + f"collection at field_path {path} contains an element of type " + f"`{type(element).__name__}` instead of an object. ForEach can only transform " + f"collections of objects. Please point `field_path` at a list of objects." + ), + internal_message=( + f"ForEach transformation received a non-object element of type " + f"{type(element).__name__} at field_path {path}" + ), + failure_type=FailureType.config_error, + ) + return cast(List[Dict[str, Any]], collection) + + def __eq__(self, other: Any) -> bool: + return bool(self.__dict__ == other.__dict__) diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 21c99adc71..bcfe3f2cb4 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -198,7 +198,7 @@ from airbyte_cdk.sources.declarative.schema.composite_schema_loader import CompositeSchemaLoader from airbyte_cdk.sources.declarative.schema.schema_loader import SchemaLoader from airbyte_cdk.sources.declarative.spec import Spec -from airbyte_cdk.sources.declarative.transformations import AddFields, RemoveFields +from airbyte_cdk.sources.declarative.transformations import AddFields, ForEach, RemoveFields from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition from airbyte_cdk.sources.declarative.transformations.keys_replace_transformation import ( KeysReplaceTransformation, @@ -3407,6 +3407,87 @@ def test_default_schema_loader(self): == f"{stream.name}.json" ) + def _get_transformations(self, content): + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + resolved_manifest["type"] = "DeclarativeSource" + stream_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["the_stream"], {} + ) + + stream = factory.create_component( + model_type=DeclarativeStreamModel, + component_definition=stream_manifest, + config=input_config, + ) + + return get_retriever(stream).record_selector.transformations + + def test_for_each(self): + content = f""" + the_stream: + type: DeclarativeStream + $parameters: + {self.base_parameters} + transformations: + - type: ForEach + field_path: ["column_values"] + transformations: + - type: AddFields + condition: "{{{{ record.get('display_value') and not record.get('text') }}}}" + fields: + - path: ["text"] + value: "{{{{ record['display_value'] }}}}" + - type: RemoveFields + field_pointers: + - ["display_value"] + """ + transformations = self._get_transformations(content) + + assert len(transformations) == 1 + for_each = transformations[0] + assert isinstance(for_each, ForEach) + assert for_each.field_path == ["column_values"] + assert len(for_each.transformations) == 2 + assert isinstance(for_each.transformations[0], AddFields) + assert isinstance(for_each.transformations[1], RemoveFields) + + record = {"column_values": [{"text": None, "display_value": "a value"}]} + for_each.transform(record, config=input_config) + assert record == {"column_values": [{"text": "a value"}]} + + def test_for_each_nested_in_for_each(self): + content = f""" + the_stream: + type: DeclarativeStream + $parameters: + {self.base_parameters} + transformations: + - type: ForEach + field_path: ["items"] + transformations: + - type: ForEach + field_path: ["column_values"] + transformations: + - type: AddFields + fields: + - path: ["added"] + value: "a value" + """ + transformations = self._get_transformations(content) + + assert len(transformations) == 1 + outer = transformations[0] + assert isinstance(outer, ForEach) + inner = outer.transformations[0] + assert isinstance(inner, ForEach) + assert inner.field_path == ["column_values"] + assert isinstance(inner.transformations[0], AddFields) + + record = {"items": [{"column_values": [{"id": 1}]}]} + outer.transform(record, config=input_config) + assert record == {"items": [{"column_values": [{"id": 1, "added": "a value"}]}]} + @pytest.mark.parametrize( "incremental, partition_router, expected_router_type, expected_stream_type", diff --git a/unit_tests/sources/declarative/transformations/test_for_each.py b/unit_tests/sources/declarative/transformations/test_for_each.py new file mode 100644 index 0000000000..8d6607b56c --- /dev/null +++ b/unit_tests/sources/declarative/transformations/test_for_each.py @@ -0,0 +1,308 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +from typing import Any, Mapping + +import pytest + +from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.transformations import AddFields, ForEach, RemoveFields +from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition +from airbyte_cdk.sources.declarative.transformations.keys_to_lower_transformation import ( + KeysToLowerTransformation, +) +from airbyte_cdk.sources.types import StreamSlice +from airbyte_cdk.utils.traced_exception import AirbyteTracedException + + +def _add_fields(path: list[str], value: str, condition: str = "") -> AddFields: + return AddFields( + fields=[ + AddedFieldDefinition(path=path, value=value, value_type=None, parameters={}), + ], + condition=condition, + parameters={}, + ) + + +def _for_each( + field_path: list[str], + transformations: list[Any], + config: Mapping[str, Any] | None = None, + parameters: Mapping[str, Any] | None = None, +) -> ForEach: + return ForEach( + config=config or {}, + field_path=field_path, + transformations=transformations, + parameters=parameters or {}, + ) + + +def test_monday_use_case_copies_display_value_into_text(): + """ + The `MondayTransformation` custom component, expressed as a ForEach. See + airbyte-integrations/connectors/source-monday/components.py:463-472 in the airbyte monorepo. + """ + transformation = _for_each( + field_path=["column_values"], + transformations=[ + _add_fields( + path=["text"], + value="{{ record['display_value'] }}", + condition="{{ record.get('display_value') and not record.get('text') }}", + ) + ], + ) + record = { + "id": 1, + "column_values": [ + {"id": 11, "text": None, "display_value": "Hola amigo!"}, + {"id": 12, "text": "already set", "display_value": "ignored"}, + {"id": 13, "text": None}, + ], + } + + transformation.transform(record) + + assert record["column_values"] == [ + {"id": 11, "text": "Hola amigo!", "display_value": "Hola amigo!"}, + {"id": 12, "text": "already set", "display_value": "ignored"}, + {"id": 13, "text": None}, + ] + + +def test_transform_returns_none(): + transformation = _for_each( + field_path=["items"], transformations=[_add_fields(["added"], "value")] + ) + + assert transformation.transform({"items": [{}]}) is None + + +def test_mutation_is_visible_on_the_parent_record_without_copying(): + transformation = _for_each( + field_path=["items"], transformations=[_add_fields(["added"], "value")] + ) + element = {"id": 1} + collection = [element] + record = {"items": collection} + + transformation.transform(record) + + # the very same objects were mutated, nothing was copied and written back + assert record["items"] is collection + assert record["items"][0] is element + assert element == {"id": 1, "added": "value"} + + +def test_nested_collection_is_reached_through_a_multi_segment_path(): + transformation = _for_each( + field_path=["data", "assets"], transformations=[_add_fields(["added"], "value")] + ) + record = {"data": {"assets": [{"id": 1}, {"id": 2}]}} + + transformation.transform(record) + + assert record == { + "data": {"assets": [{"id": 1, "added": "value"}, {"id": 2, "added": "value"}]} + } + + +def test_wildcard_segment_iterates_every_matching_collection(): + transformation = _for_each( + field_path=["data", "*", "items"], transformations=[_add_fields(["added"], "value")] + ) + record = { + "data": { + "first": {"items": [{"id": 1}]}, + "second": {"items": [{"id": 2}]}, + "third": {"other": [{"id": 3}]}, + } + } + + transformation.transform(record) + + assert record == { + "data": { + "first": {"items": [{"id": 1, "added": "value"}]}, + "second": {"items": [{"id": 2, "added": "value"}]}, + "third": {"other": [{"id": 3}]}, + } + } + + +@pytest.mark.parametrize( + "record", + [ + pytest.param({"id": 1}, id="field_path_is_absent"), + pytest.param({"id": 1, "items": None}, id="field_path_resolves_to_null"), + pytest.param({"id": 1, "items": "a string"}, id="field_path_resolves_to_a_string"), + pytest.param({"id": 1, "items": 42}, id="field_path_resolves_to_a_number"), + pytest.param({"id": 1, "items": []}, id="field_path_resolves_to_an_empty_list"), + pytest.param({"id": 1, "other": [{"a": 1}]}, id="a_sibling_field_exists"), + ], +) +def test_unresolvable_or_non_collection_field_path_is_a_no_op(record): + transformation = _for_each( + field_path=["items"], transformations=[_add_fields(["added"], "value")] + ) + expected = {key: value for key, value in record.items()} + + transformation.transform(record) + + assert record == expected + + +def test_missing_nested_field_path_is_a_no_op(): + transformation = _for_each( + field_path=["data", "items"], transformations=[_add_fields(["added"], "value")] + ) + record = {"data": {"something_else": 1}} + + transformation.transform(record) + + assert record == {"data": {"something_else": 1}} + + +def test_object_target_is_treated_as_a_collection_of_one(): + transformation = _for_each( + field_path=["item"], transformations=[_add_fields(["added"], "value")] + ) + record = {"item": {"id": 1}} + + transformation.transform(record) + + assert record == {"item": {"id": 1, "added": "value"}} + + +def test_non_object_element_raises_a_config_error_naming_the_field_path(): + transformation = _for_each( + field_path=["tags"], + transformations=[_add_fields(["added"], "value")], + parameters={"name": "the_stream"}, + ) + record = {"tags": ["a", "b"]} + + with pytest.raises(AirbyteTracedException) as exc_info: + transformation.transform(record) + + assert exc_info.value.failure_type == FailureType.config_error + assert "the_stream" in exc_info.value.message + assert "['tags']" in exc_info.value.message + assert "str" in exc_info.value.message + # nothing was mutated before the error was raised + assert record == {"tags": ["a", "b"]} + + +def test_non_object_element_error_without_a_stream_name(): + transformation = _for_each( + field_path=["tags"], transformations=[_add_fields(["added"], "value")] + ) + + with pytest.raises(AirbyteTracedException) as exc_info: + transformation.transform({"tags": [1]}) + + assert "['tags']" in exc_info.value.message + assert "int" in exc_info.value.message + + +def test_nested_transformations_receive_config_and_stream_slice(): + transformation = _for_each( + field_path=["items"], + transformations=[ + _add_fields(["shop_id"], "{{ config['shop_id'] }}"), + _add_fields(["partition_id"], "{{ stream_slice['partition_id'] }}"), + ], + ) + record = {"items": [{"id": 1}]} + + transformation.transform( + record, + config={"shop_id": "my-shop"}, + stream_slice=StreamSlice(partition={"partition_id": "p1"}, cursor_slice={}), + ) + + assert record == {"items": [{"id": 1, "shop_id": "my-shop", "partition_id": "p1"}]} + + +def test_field_path_is_interpolated(): + transformation = _for_each( + field_path=["{{ config['collection_field'] }}"], + transformations=[_add_fields(["added"], "value")], + config={"collection_field": "items"}, + ) + record = {"items": [{"id": 1}]} + + transformation.transform(record) + + assert record == {"items": [{"id": 1, "added": "value"}]} + + +def test_remove_fields_inside_the_loop(): + """ + RemoveFields mutates through `dpath.delete` rather than `dpath.new`, so it exercises a different + mutation path than AddFields. + """ + transformation = _for_each( + field_path=["assets"], + transformations=[ + _add_fields(["uploader_id"], "{{ record['uploader']['id'] }}"), + RemoveFields(field_pointers=[["uploader"]], parameters={}), + ], + ) + record = {"assets": [{"id": 1, "uploader": {"id": 7}}, {"id": 2, "uploader": {"id": 8}}]} + + transformation.transform(record) + + assert record == {"assets": [{"id": 1, "uploader_id": 7}, {"id": 2, "uploader_id": 8}]} + + +def test_transformation_that_ignores_the_config_still_works_inside_the_loop(): + transformation = _for_each(field_path=["items"], transformations=[KeysToLowerTransformation()]) + record = {"items": [{"ID": 1}, {"Name": "a"}]} + + transformation.transform(record) + + assert record == {"items": [{"id": 1}, {"name": "a"}]} + + +def test_for_each_nested_inside_for_each(): + inner = _for_each( + field_path=["column_values"], + transformations=[_add_fields(["added"], "{{ config['marker'] }}")], + ) + outer = _for_each(field_path=["items"], transformations=[inner]) + record = { + "items": [ + {"id": 1, "column_values": [{"c": 1}, {"c": 2}]}, + {"id": 2, "column_values": [{"c": 3}]}, + {"id": 3}, + ] + } + + outer.transform(record, config={"marker": "x"}) + + assert record == { + "items": [ + {"id": 1, "column_values": [{"c": 1, "added": "x"}, {"c": 2, "added": "x"}]}, + {"id": 2, "column_values": [{"c": 3, "added": "x"}]}, + {"id": 3}, + ] + } + + +def test_multiple_nested_transformations_are_applied_in_order(): + transformation = _for_each( + field_path=["items"], + transformations=[ + _add_fields(["first"], "a"), + _add_fields(["second"], "{{ record['first'] }}b"), + ], + ) + record = {"items": [{}]} + + transformation.transform(record) + + assert record == {"items": [{"first": "a", "second": "ab"}]} From b11b3218d981afe2fd89b7b5e3d06e81a4d08998 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 15 Sep 2026 20:14:08 +0300 Subject: [PATCH 2/4] fix: address review findings on the ForEach transformation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Behaviour - A `null` inside the collection is now skipped instead of aborting the sync. A null in an array is ordinary API payload, and `MondayTransformation` — the Python code this component replaces — skips it, so raising was a regression. Any other non-object element still raises, but as `system_error` rather than `config_error`: a mis-pointed `field_path` is a connector-development fault the end user cannot fix from their connection configuration, and sibling decoders already classify read-time payload-shape faults that way. - Every collection a glob expands to is resolved and validated before any element is transformed, so the documented "a bad element leaves the record untouched" contract now holds per record, not just per collection. - A `field_path` segment that interpolates to something other than a non-empty string is now a `config_error` instead of silently turning the whole transformation into a no-op for the entire sync. - The error message no longer promises a stream name. `name` is a `DeclarativeStream` field, not a `$parameter`, so `self._parameters.get("name")` was always empty for a modern manifest. Threading the real name through `_create_component_from_model` was rejected: the kwarg would reach `create_custom_component`, which merges unknown kwargs into the custom class's own fields and would override a manifest-declared `name` on any existing custom transformation. The message still names the `field_path` and the element type. - Messages shortened and de-duplicated per the error-message guidelines. Performance - `dpath.get` resolves by folding over the entire object graph, so it costs O(total nodes in the record) per record: 17.7 ms measured on a record with a 10,000-element *unrelated* sibling. Paths without a glob segment now use a plain walk — 0.4 us on the same record, ~45,000x faster. `dpath.values` is used for every glob form, which also removes the bare `ValueError: globs must match only one leaf` that `**`, `?` and `a[ab]` used to raise. Manifest contract - `"ForEach.transformations": "CustomTransformation"` added to `CUSTOM_COMPONENTS_MAPPING`. Without it a `class_name`-only custom transformation — the style used in six production manifests — gets no `type` injected when nested in a `ForEach` and fails schema validation. - `field_path` declares `interpolation_context: [config, parameters]`, matching the examples the block advertises. The generated model mirrors the reworded description exactly; `interpolation_context` is not emitted by codegen. - The redundant `__eq__` override is gone. The base class body is byte-identical, and the dataclass-generated one correctly returns False for foreign types instead of raising `AttributeError`. Tests - Null elements skipped; non-object elements raise `system_error`; no stream name in the message; nothing mutated when a later collection under a glob is invalid (the previous assertion was vacuous — every element in its fixture was a non-dict, so nothing could have mutated). Both new guards were confirmed to fail against the pre-fix implementation. - `field_path: []`, `$parameters` interpolation of `field_path`, `stream_state` forwarding, non-`*` globs, equality, and a deep copy in the no-op test. - `dpath` is not called for a path without a glob, plus a wall-clock guard that a large unrelated sibling does not dominate the lookup. - A `ForEach` manifest fragment is validated against the JSON schema, including all four `$ref` sites — `DeclarativeStream.transformations`, `DynamicSchemaLoader.schema_transformations`, `JsonSchemaPropertySelector.transformations` and `ForEach.transformations`. Nothing validated against the schema before, so dropping `ForEach` from one of those `anyOf` lists would only have surfaced in a real connector. - A factory-built nested `class_name`-only `CustomTransformation`. Notes for the PR body - Do not run `/poe build` on this branch. The hand-applied model delta was reproduced against pinned codegen and is byte-identical to it; `main`'s generated file is itself ~240 lines divergent from codegen (including an `OAuthScope` -> `Scope`/`OptionalScope` public rename), so regenerating here would import that drift and would make #1162 conflict. The drift belongs in a separate re-sync PR against `main`. - The earlier `ForwardRef` rationale for the class placement was wrong. Codegen also emits `ForEach` before `JsonSchemaPropertySelector`; there is no hazard and CI regeneration would not reintroduce the failure. - The `source-github` `ReleasesRecordTransformation` citation is from the unmerged `tolik0/source-github/graphql-streams` branch, not monorepo `main`. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 5 +- .../models/declarative_component_schema.py | 2 +- .../parsers/manifest_component_transformer.py | 1 + .../declarative/transformations/for_each.py | 119 ++++-- .../test_model_to_component_factory.py | 30 ++ .../declarative/parsers/testing_components.py | 23 ++ .../transformations/test_for_each.py | 347 +++++++++++++++++- 7 files changed, 473 insertions(+), 54 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 4d55e5235f..142dab2b5a 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -3020,10 +3020,13 @@ definitions: enum: [ForEach] field_path: title: Field Path - description: Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. A `*` wildcard segment is supported, in which case every matching value is iterated over. An element that is not an object raises an error, because it cannot be transformed in place. + description: Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. Glob segments (`*`, `?`, `[...]`) are supported, in which case every matching value is iterated over. A `null` element inside the collection is skipped, because a null in an array is ordinary payload; any other non-object element raises an error, because it cannot be transformed in place. type: array items: type: string + interpolation_context: + - config + - parameters examples: - ["column_values"] - ["data", "assets"] diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 5ad4938070..cc040bf637 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -2335,7 +2335,7 @@ class ForEach(BaseModel): type: Literal["ForEach"] field_path: List[str] = Field( ..., - description="Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. A `*` wildcard segment is supported, in which case every matching value is iterated over. An element that is not an object raises an error, because it cannot be transformed in place.", + description="Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. Glob segments (`*`, `?`, `[...]`) are supported, in which case every matching value is iterated over. A `null` element inside the collection is skipped, because a null in an array is ordinary payload; any other non-object element raises an error, because it cannot be transformed in place.", examples=[ ["column_values"], ["data", "assets"], diff --git a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py index 00b1a18ffb..df7fa28bd8 100644 --- a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py +++ b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py @@ -77,6 +77,7 @@ "DeclarativeStream.transformations": "CustomTransformation", "DefaultErrorHandler.backoff_strategies": "CustomBackoffStrategy", "DefaultPaginator.pagination_strategy": "CustomPaginationStrategy", + "ForEach.transformations": "CustomTransformation", "HttpRequester.authenticator": "CustomAuthenticator", "HttpRequester.error_handler": "CustomErrorHandler", "RecordSelector.extractor": "CustomRecordExtractor", diff --git a/airbyte_cdk/sources/declarative/transformations/for_each.py b/airbyte_cdk/sources/declarative/transformations/for_each.py index 3ad89fc94c..2affade5ac 100644 --- a/airbyte_cdk/sources/declarative/transformations/for_each.py +++ b/airbyte_cdk/sources/declarative/transformations/for_each.py @@ -3,7 +3,7 @@ # from dataclasses import InitVar, dataclass -from typing import Any, Dict, List, Mapping, Optional, Union, cast +from typing import Any, Dict, List, Mapping, Optional, Union import dpath @@ -13,6 +13,11 @@ from airbyte_cdk.sources.types import Config, StreamSlice, StreamState from airbyte_cdk.utils.traced_exception import AirbyteTracedException +# Segments containing any of these need `dpath` glob matching. Everything else is resolved by a +# plain walk, because `dpath` traverses the whole record to glob-match and that cost is paid per +# record. +_GLOB_CHARACTERS = ("*", "?", "[") + @dataclass class ForEach(RecordTransformation): @@ -25,9 +30,11 @@ class ForEach(RecordTransformation): be written back. Behavior of `field_path` resolution: - * The path entries are interpolated strings, like every other dpath-based component. - * A `*` wildcard segment is supported. When the path contains a `*`, every value matching the glob is - treated as its own collection to iterate over. + * The path entries are interpolated strings, like every other dpath-based component. A segment that + interpolates to something other than a non-empty string is a configuration error, because it would + otherwise turn the whole transformation into a silent no-op. + * Glob segments (`*`, `?`, `[...]`) are supported. When the path contains one, every value matching the + glob is treated as its own collection to iterate over. * If the path does not resolve, the transformation is a no-op. A collection missing from some records is expected and must not fail the sync. * If the resolved value is a list, the nested transformations are applied to each element. @@ -36,9 +43,13 @@ class ForEach(RecordTransformation): * If the resolved value is a scalar or `None`, the transformation is a no-op. * An empty `field_path` resolves to the record itself, the same way `DpathExtractor` treats an empty `field_path`. - * If an element of the resolved list is not an object (for example a list of strings), an error is - raised. Such an element cannot be mutated in place, and silently skipping it would hide a manifest - mistake behind missing data. + * A `null` element inside the collection is skipped. A `null` in an array is ordinary API payload, not a + manifest mistake, and the Python transformations this component replaces skip it too. + * Any other non-object element (a string, a number) raises. Such an element cannot be mutated in place, + and silently skipping it would hide a mis-pointed `field_path` behind missing data. + + Every collection the path matches is resolved and validated before any element is transformed, so a bad + element anywhere under a glob leaves the whole record untouched. Example: ```yaml @@ -78,56 +89,92 @@ def transform( stream_slice: Optional[StreamSlice] = None, ) -> None: effective_config = config if config is not None else self.config - path = [path.eval(effective_config) for path in self._field_path] - - for collection in self._resolve_collections(record, path): - for element in self._iterate(collection, path): - for transformation in self.transformations: - transformation.transform( - element, - config=effective_config, - stream_state=stream_state, - stream_slice=stream_slice, - ) + path = self._evaluate_path(effective_config) + + # Resolving and validating every matched collection up front is what makes the documented + # "a bad element leaves the record untouched" contract hold across a glob, not just within a + # single collection. + elements = [ + element + for collection in self._resolve_collections(record, path) + for element in self._elements_of(collection, path) + ] + + for element in elements: + for transformation in self.transformations: + transformation.transform( + element, + config=effective_config, + stream_state=stream_state, + stream_slice=stream_slice, + ) + + def _evaluate_path(self, config: Config) -> List[str]: + path = [] + for interpolated_segment in self._field_path: + segment = interpolated_segment.eval(config) + if not isinstance(segment, str) or not segment: + raise AirbyteTracedException( + message=( + f"ForEach cannot resolve `field_path` {self.field_path}: a path segment " + f"evaluated to {segment!r} instead of a non-empty string. Check the " + f"configuration values it interpolates." + ), + internal_message=( + f"ForEach field_path segment evaluated to {segment!r} " + f"({type(segment).__name__}) for field_path {self.field_path}" + ), + failure_type=FailureType.config_error, + ) + path.append(segment) + return path @staticmethod def _resolve_collections(record: Dict[str, Any], path: List[str]) -> List[Any]: """ - Resolve `path` to the collections to iterate over. `dpath` returns references to the nested objects, - never copies, which is what makes the in-place mutation work. + Resolve `path` to the collections to iterate over. Both the plain walk and `dpath` return + references to the nested objects, never copies, which is what makes the in-place mutation work. """ if not path: return [record] - if "*" in path: + if any(character in segment for segment in path for character in _GLOB_CHARACTERS): return list(dpath.values(record, path)) - resolved = dpath.get(record, path, default=None) - return [] if resolved is None else [resolved] - def _iterate(self, collection: Any, path: List[str]) -> List[Dict[str, Any]]: + node: Any = record + for segment in path: + if isinstance(node, dict): + if segment not in node: + return [] + node = node[segment] + elif isinstance(node, list) and segment.isdigit() and int(segment) < len(node): + node = node[int(segment)] + else: + return [] + return [] if node is None else [node] + + @staticmethod + def _elements_of(collection: Any, path: List[str]) -> List[Dict[str, Any]]: if isinstance(collection, dict): return [collection] if not isinstance(collection, list): # A scalar cannot hold nested records, so there is nothing to transform. return [] + elements = [] for element in collection: - if not isinstance(element, dict): - stream_name = self._parameters.get("name") - stream_context = f" of stream `{stream_name}`" if stream_name else "" + if isinstance(element, dict): + elements.append(element) + elif element is not None: raise AirbyteTracedException( message=( - f"The ForEach transformation{stream_context} could not be applied because the " - f"collection at field_path {path} contains an element of type " - f"`{type(element).__name__}` instead of an object. ForEach can only transform " - f"collections of objects. Please point `field_path` at a list of objects." + f"ForEach cannot transform the collection at field_path {path}: it holds a " + f"non-object element ({type(element).__name__}). Set `field_path` to a list " + f"of objects." ), internal_message=( f"ForEach transformation received a non-object element of type " f"{type(element).__name__} at field_path {path}" ), - failure_type=FailureType.config_error, + failure_type=FailureType.system_error, ) - return cast(List[Dict[str, Any]], collection) - - def __eq__(self, other: Any) -> bool: - return bool(self.__dict__ == other.__dict__) + return elements diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index bcfe3f2cb4..200f0bddbe 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -226,6 +226,7 @@ from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse from unit_tests.sources.declarative.parsers.testing_components import ( TestingCustomSubstreamPartitionRouter, + TestingCustomTransformation, TestingSomeComponent, ) @@ -3488,6 +3489,35 @@ def test_for_each_nested_in_for_each(self): outer.transform(record, config=input_config) assert record == {"items": [{"column_values": [{"id": 1, "added": "a value"}]}]} + def test_for_each_with_a_class_name_only_custom_transformation(self): + """ + `class_name`-only custom transformations work at stream level thanks to + CUSTOM_COMPONENTS_MAPPING. `ForEach.transformations` needs its own entry or the nested + component never gets a `type` and fails to parse. + """ + content = f""" + the_stream: + type: DeclarativeStream + $parameters: + {self.base_parameters} + transformations: + - type: ForEach + field_path: ["items"] + transformations: + - class_name: unit_tests.sources.declarative.parsers.testing_components.TestingCustomTransformation + marker: "from the manifest" + """ + transformations = self._get_transformations(content) + + assert len(transformations) == 1 + for_each = transformations[0] + assert isinstance(for_each, ForEach) + assert isinstance(for_each.transformations[0], TestingCustomTransformation) + + record = {"items": [{"id": 1}]} + for_each.transform(record, config=input_config) + assert record == {"items": [{"id": 1, "marker": "from the manifest"}]} + @pytest.mark.parametrize( "incremental, partition_router, expected_router_type, expected_stream_type", diff --git a/unit_tests/sources/declarative/parsers/testing_components.py b/unit_tests/sources/declarative/parsers/testing_components.py index b21b63efd6..74d86f2998 100644 --- a/unit_tests/sources/declarative/parsers/testing_components.py +++ b/unit_tests/sources/declarative/parsers/testing_components.py @@ -21,6 +21,8 @@ RequestInput, ) from airbyte_cdk.sources.declarative.retrievers import SimpleRetriever +from airbyte_cdk.sources.declarative.transformations import RecordTransformation +from airbyte_cdk.sources.types import Config, StreamSlice, StreamState @dataclass @@ -121,3 +123,24 @@ class TestingRequesterWithDefaultBudget(TestingRequester): def __post_init__(self, parameters: Mapping[str, Any]) -> None: self.api_budget = None super().__post_init__(parameters) + + +@dataclass +class TestingCustomTransformation(RecordTransformation): + """ + A test class used for manifests that declare a custom transformation by `class_name` only. + """ + + __test__: ClassVar[bool] = False # Tell Pytest this is not a Pytest class, despite its name + + config: Config = field(default_factory=dict) + marker: str = "custom" + + def transform( + self, + record: Mapping[str, Any], + config: Optional[Config] = None, + stream_state: Optional[StreamState] = None, + stream_slice: Optional[StreamSlice] = None, + ) -> None: + record["marker"] = self.marker diff --git a/unit_tests/sources/declarative/transformations/test_for_each.py b/unit_tests/sources/declarative/transformations/test_for_each.py index 8d6607b56c..1cc32f944c 100644 --- a/unit_tests/sources/declarative/transformations/test_for_each.py +++ b/unit_tests/sources/declarative/transformations/test_for_each.py @@ -2,12 +2,24 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. # +import copy +import time from typing import Any, Mapping +from unittest.mock import patch import pytest +from jsonschema import ValidationError, validate from airbyte_cdk.models import FailureType -from airbyte_cdk.sources.declarative.transformations import AddFields, ForEach, RemoveFields +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + _get_declarative_component_schema, +) +from airbyte_cdk.sources.declarative.transformations import ( + AddFields, + ForEach, + RecordTransformation, + RemoveFields, +) from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition from airbyte_cdk.sources.declarative.transformations.keys_to_lower_transformation import ( KeysToLowerTransformation, @@ -148,7 +160,7 @@ def test_unresolvable_or_non_collection_field_path_is_a_no_op(record): transformation = _for_each( field_path=["items"], transformations=[_add_fields(["added"], "value")] ) - expected = {key: value for key, value in record.items()} + expected = copy.deepcopy(record) transformation.transform(record) @@ -177,35 +189,118 @@ def test_object_target_is_treated_as_a_collection_of_one(): assert record == {"item": {"id": 1, "added": "value"}} -def test_non_object_element_raises_a_config_error_naming_the_field_path(): +def test_null_element_inside_the_collection_is_skipped(): + """ + A `null` in an array is ordinary API payload, not a manifest mistake. The Python transformations + this component replaces (for example `MondayTransformation`) skip it, so ForEach must too. + """ + transformation = _for_each( + field_path=["items"], transformations=[_add_fields(["added"], "value")] + ) + record = {"items": [{"id": 1}, None, {"id": 2}]} + + transformation.transform(record) + + assert record == {"items": [{"id": 1, "added": "value"}, None, {"id": 2, "added": "value"}]} + + +def test_collection_of_only_nulls_is_a_no_op(): + transformation = _for_each( + field_path=["items"], transformations=[_add_fields(["added"], "value")] + ) + record = {"items": [None, None]} + + transformation.transform(record) + + assert record == {"items": [None, None]} + + +@pytest.mark.parametrize( + "collection, expected_type_name", + [ + pytest.param(["a", "b"], "str", id="strings"), + pytest.param([1], "int", id="ints"), + pytest.param([[{"id": 1}]], "list", id="nested_lists"), + ], +) +def test_non_object_element_raises_a_system_error_naming_the_field_path( + collection, expected_type_name +): + """ + Unlike a `null`, a string or a number in the collection means `field_path` points at the wrong + thing. That is a connector-development fault, so it is a `system_error` rather than something the + user is told to fix in their connection configuration. + """ + transformation = _for_each( + field_path=["tags"], transformations=[_add_fields(["added"], "value")] + ) + record = {"tags": collection} + + with pytest.raises(AirbyteTracedException) as exc_info: + transformation.transform(record) + + assert exc_info.value.failure_type == FailureType.system_error + assert "['tags']" in exc_info.value.message + assert expected_type_name in exc_info.value.message + assert record == {"tags": collection} + + +def test_the_error_message_does_not_promise_a_stream_name(): + """ + `name` is a DeclarativeStream field, not a `$parameter`, so a stream name sourced from + `$parameters` never materialises for a modern manifest. The message must not claim one. + """ transformation = _for_each( field_path=["tags"], transformations=[_add_fields(["added"], "value")], parameters={"name": "the_stream"}, ) - record = {"tags": ["a", "b"]} with pytest.raises(AirbyteTracedException) as exc_info: - transformation.transform(record) + transformation.transform({"tags": ["a"]}) - assert exc_info.value.failure_type == FailureType.config_error - assert "the_stream" in exc_info.value.message - assert "['tags']" in exc_info.value.message - assert "str" in exc_info.value.message - # nothing was mutated before the error was raised - assert record == {"tags": ["a", "b"]} + assert "the_stream" not in exc_info.value.message + assert "stream" not in exc_info.value.message -def test_non_object_element_error_without_a_stream_name(): +def test_nothing_is_mutated_when_a_later_element_of_the_same_collection_is_invalid(): transformation = _for_each( field_path=["tags"], transformations=[_add_fields(["added"], "value")] ) + record = {"tags": [{"id": 1}, "bad"]} - with pytest.raises(AirbyteTracedException) as exc_info: - transformation.transform({"tags": [1]}) + with pytest.raises(AirbyteTracedException): + transformation.transform(record) - assert "['tags']" in exc_info.value.message - assert "int" in exc_info.value.message + assert record == {"tags": [{"id": 1}, "bad"]} + + +def test_nothing_is_mutated_when_a_later_collection_under_a_wildcard_is_invalid(): + """ + The whole-record pre-pass has to cover every collection the glob expands to, not just the one + currently being iterated, or the record is left half-transformed. + """ + transformation = _for_each( + field_path=["data", "*", "items"], transformations=[_add_fields(["added"], "value")] + ) + record = { + "data": { + "first": {"items": [{"id": 1}]}, + "second": {"items": [{"id": 2}]}, + "third": {"items": ["bad"]}, + } + } + + with pytest.raises(AirbyteTracedException): + transformation.transform(record) + + assert record == { + "data": { + "first": {"items": [{"id": 1}]}, + "second": {"items": [{"id": 2}]}, + "third": {"items": ["bad"]}, + } + } def test_nested_transformations_receive_config_and_stream_slice(): @@ -306,3 +401,223 @@ def test_multiple_nested_transformations_are_applied_in_order(): transformation.transform(record) assert record == {"items": [{"first": "a", "second": "ab"}]} + + +def test_empty_field_path_applies_the_transformations_to_the_record_itself(): + transformation = _for_each(field_path=[], transformations=[_add_fields(["added"], "value")]) + record = {"id": 1} + + transformation.transform(record) + + assert record == {"id": 1, "added": "value"} + + +class _RecordingTransformation(RecordTransformation): + def __init__(self): + self.calls = [] + + def transform(self, record, config=None, stream_state=None, stream_slice=None) -> None: + self.calls.append( + { + "record": record, + "config": config, + "stream_state": stream_state, + "stream_slice": stream_slice, + } + ) + + +def test_stream_state_and_stream_slice_are_forwarded_to_the_nested_transformations(): + recorder = _RecordingTransformation() + transformation = _for_each(field_path=["items"], transformations=[recorder], config={"a": 1}) + element = {"id": 1} + stream_slice = StreamSlice(partition={"partition_id": "p1"}, cursor_slice={}) + + transformation.transform( + {"items": [element]}, stream_state={"cursor": "2024-01-01"}, stream_slice=stream_slice + ) + + assert recorder.calls == [ + { + "record": element, + "config": {"a": 1}, + "stream_state": {"cursor": "2024-01-01"}, + "stream_slice": stream_slice, + } + ] + + +def test_field_path_is_interpolated_from_parameters(): + transformation = _for_each( + field_path=["{{ parameters['collection_field'] }}"], + transformations=[_add_fields(["added"], "value")], + parameters={"collection_field": "items"}, + ) + record = {"items": [{"id": 1}]} + + transformation.transform(record) + + assert record == {"items": [{"id": 1, "added": "value"}]} + + +@pytest.mark.parametrize( + "config", + [ + pytest.param({}, id="segment_interpolates_to_an_empty_string"), + pytest.param({"collection_field": 5}, id="segment_interpolates_to_a_number"), + ], +) +def test_a_field_path_segment_that_is_not_a_non_empty_string_is_a_config_error(config): + """ + Without this check the component is a silent no-op for the entire sync, which is the hardest + failure mode to diagnose. + """ + transformation = _for_each( + field_path=["{{ config['collection_field'] }}"], + transformations=[_add_fields(["added"], "value")], + config=config, + ) + + with pytest.raises(AirbyteTracedException) as exc_info: + transformation.transform({"items": [{"id": 1}]}) + + assert exc_info.value.failure_type == FailureType.config_error + + +@pytest.mark.parametrize( + "field_path", + [ + pytest.param(["**", "items"], id="recursive_glob"), + pytest.param(["data", "?", "items"], id="single_character_glob"), + pytest.param(["data", "item*"], id="partial_glob"), + ], +) +def test_non_star_globs_match_several_collections_instead_of_raising(field_path): + """ + `dpath.get` raises a bare `ValueError` as soon as a glob matches more than one leaf. Routing every + glob through `dpath.values` keeps the multi-match case working. + """ + transformation = _for_each( + field_path=field_path, transformations=[_add_fields(["added"], "value")] + ) + record = {"data": {"a": {"items": [{"id": 1}]}, "items": [{"id": 2}], "itemz": [{"id": 3}]}} + + transformation.transform(record) + + assert any( + element.get("added") == "value" + for collection in ( + record["data"]["a"]["items"], + record["data"]["items"], + record["data"]["itemz"], + ) + for element in collection + ) + + +def test_a_path_without_a_glob_never_goes_through_dpath(): + """ + `dpath` resolves by folding over the whole object graph, so it costs O(total nodes in the record) + per call. The plain-walk fast path is what keeps a large unrelated sibling from dominating. + """ + transformation = _for_each( + field_path=["data", "items"], transformations=[_add_fields(["added"], "value")] + ) + record = {"data": {"items": [{"id": 1}]}, "junk": [{"n": n} for n in range(100)]} + + with patch("airbyte_cdk.sources.declarative.transformations.for_each.dpath") as mocked_dpath: + transformation.transform(record) + + mocked_dpath.get.assert_not_called() + mocked_dpath.values.assert_not_called() + assert record["data"]["items"] == [{"id": 1, "added": "value"}] + + +def test_a_large_unrelated_sibling_does_not_slow_down_the_lookup(): + transformation = _for_each( + field_path=["items"], transformations=[_add_fields(["added"], "value")] + ) + junk = [{"n": n} for n in range(20_000)] + + started_at = time.perf_counter() + for _ in range(500): + transformation.transform({"items": [{"id": 1}], "junk": junk}) + elapsed = time.perf_counter() - started_at + + # Resolving through `dpath.get` takes ~35 ms per record here, so the same loop takes ~17 s. + assert elapsed < 2.0, ( + f"resolving `field_path` scaled with the unrelated sibling: {elapsed:.2f}s" + ) + + +def test_equality_compares_the_configured_fields_and_tolerates_foreign_types(): + def build(field_path): + return _for_each(field_path=field_path, transformations=[_add_fields(["added"], "value")]) + + assert build(["items"]) == build(["items"]) + assert build(["items"]) != build(["other"]) + assert build(["items"]) != 1 + + +_FOR_EACH_MANIFEST_FRAGMENT = { + "type": "ForEach", + "field_path": ["column_values"], + "transformations": [ + { + "type": "AddFields", + "fields": [{"type": "AddedFieldDefinition", "path": ["text"], "value": "a value"}], + }, + { + "type": "ForEach", + "field_path": ["nested"], + "transformations": [{"type": "RemoveFields", "field_pointers": [["uploader"]]}], + }, + {"type": "CustomTransformation", "class_name": "source_x.components.MyTransformation"}, + ], +} + + +def _definition_schema(definition_name: str) -> dict[str, Any]: + schema = _get_declarative_component_schema() + # The nested `$ref`s are all `#/definitions/...`, so they resolve as long as `definitions` sits + # at the root of the schema handed to `validate`. + return {**schema["definitions"][definition_name], "definitions": schema["definitions"]} + + +def test_a_for_each_component_validates_against_the_declarative_component_schema(): + validate(instance=_FOR_EACH_MANIFEST_FRAGMENT, schema=_definition_schema("ForEach")) + + +@pytest.mark.parametrize( + "definition_name, transformations_field", + [ + pytest.param("DeclarativeStream", "transformations", id="declarative_stream"), + pytest.param("DynamicSchemaLoader", "schema_transformations", id="dynamic_schema_loader"), + pytest.param( + "JsonSchemaPropertySelector", "transformations", id="json_schema_property_selector" + ), + pytest.param("ForEach", "transformations", id="for_each"), + ], +) +def test_for_each_is_accepted_by_every_transformations_slot(definition_name, transformations_field): + """ + Regression guard for the four `$ref` sites. `create_component` never validates against the JSON + schema, so dropping `ForEach` from one of these `anyOf` lists would otherwise only surface in a + real connector. + """ + schema = _get_declarative_component_schema() + property_schema = { + **schema["definitions"][definition_name]["properties"][transformations_field], + "definitions": schema["definitions"], + } + + validate(instance=[_FOR_EACH_MANIFEST_FRAGMENT], schema=property_schema) + + +def test_a_for_each_without_a_field_path_is_rejected_by_the_schema(): + invalid = { + key: value for key, value in _FOR_EACH_MANIFEST_FRAGMENT.items() if key != "field_path" + } + + with pytest.raises(ValidationError): + validate(instance=invalid, schema=_definition_schema("ForEach")) From 7c4a4d49b8ca931f16c68a5d235a65b69fcc6298 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 15 Sep 2026 20:43:44 +0300 Subject: [PATCH 3/4] fix: restore dpath parity in the ForEach fast path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The round-2 performance fix replaced `dpath.get` with a hand-rolled walk for non-glob paths. The walk is worth keeping -- it takes path resolution from 18 ms to 0.35 us per record -- but it diverged from `dpath` in two ways, and a third divergence was introduced alongside it. * Negative list indices were dropped. `"-1".isdigit()` is `False`, so the walk fell through to `return []` where `dpath` returns the last element. * `isdigit()` is the wrong predicate: it is `True` for superscripts such as `"²"` that `int()` rejects, so such a segment raised an uncaught `ValueError` against a list node, where the contract is a no-op. `dpath` decides with `int()` itself (`dpath.segments.match`, through `ListIndex.__eq__`), so the walk now does too -- which also covers the `+1`, `" 1"` and `"1_0"` forms that `isdecimal()` would have rejected. * Neither was reachable in the first place. `_evaluate_path`, added in the same round, rejects any segment that is not a string, and Jinja renders `"-1"` as an `int` -- so every list index, `"0"` as much as `"-1"`, was a `config_error` before the walk ever ran. Integer segments are now accepted and stringified; `dpath` takes an integer list index too. Also: a literal empty `field_path` segment was classified `config_error`, but an empty string cannot be the render of a template, so it is a manifest fault. It is now a `system_error` raised once at construction, consistent with how `_elements_of` classifies a mis-pointed `field_path`. Drops the now-dead `self._parameters`, matching `AddFields`/`RemoveFields`. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative/transformations/for_each.py | 47 +++++++- .../transformations/test_for_each.py | 112 +++++++++++++++++- 2 files changed, 153 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/declarative/transformations/for_each.py b/airbyte_cdk/sources/declarative/transformations/for_each.py index 2affade5ac..8d41b3d05e 100644 --- a/airbyte_cdk/sources/declarative/transformations/for_each.py +++ b/airbyte_cdk/sources/declarative/transformations/for_each.py @@ -19,6 +19,20 @@ _GLOB_CHARACTERS = ("*", "?", "[") +def _as_list_index(segment: str, length: int) -> Optional[int]: + """ + Resolve `segment` against a list of `length` elements exactly the way `dpath` does: `int()` is + what decides whether a segment is an index at all (`dpath.segments.match`), and a negative index + counts from the end (`dpath.types.ListIndex.__eq__`). A segment `int()` rejects, or an index + outside the list, matches nothing -- the empty result `dpath.values` returns. + """ + try: + index = int(segment) + except ValueError: + return None + return index if -length <= index < length else None + + @dataclass class ForEach(RecordTransformation): """ @@ -31,8 +45,11 @@ class ForEach(RecordTransformation): Behavior of `field_path` resolution: * The path entries are interpolated strings, like every other dpath-based component. A segment that - interpolates to something other than a non-empty string is a configuration error, because it would - otherwise turn the whole transformation into a silent no-op. + interpolates to something other than a non-empty string or an integer is a configuration error, + because it would otherwise turn the whole transformation into a silent no-op. An empty segment + written literally in the manifest is rejected at construction time instead, as a manifest error. + * A segment addressing a list is resolved the way `dpath` resolves it: any value `int()` accepts is + an index, and a negative index counts from the end. An index outside the list matches nothing. * Glob segments (`*`, `?`, `[...]`) are supported. When the path contains one, every value matching the glob is treated as its own collection to iterate over. * If the path does not resolve, the transformation is a no-op. A collection missing from some records is @@ -76,10 +93,23 @@ class ForEach(RecordTransformation): parameters: InitVar[Mapping[str, Any]] def __post_init__(self, parameters: Mapping[str, Any]) -> None: - self._parameters = parameters self._field_path: List[InterpolatedString] = [ InterpolatedString.create(path, parameters=parameters) for path in self.field_path ] + # An empty segment cannot be a template, so it can only come from the manifest. Catching it + # here rather than per record keeps it classified as the manifest error it is. + if any(not segment.string for segment in self._field_path): + raise AirbyteTracedException( + message=( + f"ForEach cannot resolve `field_path` {self.field_path}: it contains an empty " + f"path segment. Remove the empty segment, or set `field_path` to `[]` to target " + f"the record itself." + ), + internal_message=( + f"ForEach field_path contains an empty literal segment: {self.field_path}" + ), + failure_type=FailureType.system_error, + ) def transform( self, @@ -113,6 +143,10 @@ def _evaluate_path(self, config: Config) -> List[str]: path = [] for interpolated_segment in self._field_path: segment = interpolated_segment.eval(config) + # Jinja renders a numeric segment as an `int`, and `dpath` accepts an integer list + # index, so a number is a valid path element rather than a broken one. + if isinstance(segment, int) and not isinstance(segment, bool): + segment = str(segment) if not isinstance(segment, str) or not segment: raise AirbyteTracedException( message=( @@ -146,8 +180,11 @@ def _resolve_collections(record: Dict[str, Any], path: List[str]) -> List[Any]: if segment not in node: return [] node = node[segment] - elif isinstance(node, list) and segment.isdigit() and int(segment) < len(node): - node = node[int(segment)] + elif isinstance(node, list): + index = _as_list_index(segment, len(node)) + if index is None: + return [] + node = node[index] else: return [] return [] if node is None else [node] diff --git a/unit_tests/sources/declarative/transformations/test_for_each.py b/unit_tests/sources/declarative/transformations/test_for_each.py index 1cc32f944c..9403f289a9 100644 --- a/unit_tests/sources/declarative/transformations/test_for_each.py +++ b/unit_tests/sources/declarative/transformations/test_for_each.py @@ -7,6 +7,7 @@ from typing import Any, Mapping from unittest.mock import patch +import dpath import pytest from jsonschema import ValidationError, validate @@ -464,7 +465,10 @@ def test_field_path_is_interpolated_from_parameters(): "config", [ pytest.param({}, id="segment_interpolates_to_an_empty_string"), - pytest.param({"collection_field": 5}, id="segment_interpolates_to_a_number"), + pytest.param({"collection_field": 1.5}, id="segment_interpolates_to_a_float"), + pytest.param({"collection_field": True}, id="segment_interpolates_to_a_boolean"), + pytest.param({"collection_field": None}, id="segment_interpolates_to_null"), + pytest.param({"collection_field": ["items"]}, id="segment_interpolates_to_a_list"), ], ) def test_a_field_path_segment_that_is_not_a_non_empty_string_is_a_config_error(config): @@ -515,6 +519,112 @@ def test_non_star_globs_match_several_collections_instead_of_raising(field_path) ) +@pytest.mark.parametrize( + "index, expected_position", + [ + pytest.param("-1", 2, id="minus_one_is_the_last_element"), + pytest.param("-2", 1, id="minus_two_is_the_second_to_last_element"), + pytest.param("-3", 0, id="minus_three_is_the_first_element"), + pytest.param("-0", 0, id="minus_zero_is_the_first_element"), + pytest.param("0", 0, id="zero_is_the_first_element"), + pytest.param("2", 2, id="a_positive_index_still_works"), + ], +) +def test_a_list_index_segment_selects_a_single_element(index, expected_position): + """ + `dpath` supports negative list indices, so the plain-walk fast path has to as well, or the same + `field_path` resolves differently depending on whether the path happens to contain a glob. + """ + transformation = _for_each( + field_path=["a", index], transformations=[_add_fields(["added"], "value")] + ) + record = {"a": [{"id": 0}, {"id": 1}, {"id": 2}]} + + transformation.transform(record) + + assert record["a"][expected_position] == {"id": expected_position, "added": "value"} + assert [element for element in record["a"] if "added" in element] == [ + {"id": expected_position, "added": "value"} + ] + + +@pytest.mark.parametrize( + "segment", + [ + pytest.param("-4", id="negative_index_past_the_start"), + pytest.param("3", id="positive_index_past_the_end"), + pytest.param("\u00b2", id="superscript_two_is_a_digit_that_int_rejects"), + pytest.param("\u2155", id="vulgar_fraction_is_numeric_but_not_an_index"), + pytest.param("abc", id="a_plain_word_against_a_list"), + ], +) +def test_a_segment_that_is_not_an_in_range_list_index_is_a_no_op(segment): + """ + `"\u00b2".isdigit()` is `True` but `int("\u00b2")` raises, so classifying list segments with + `isdigit()` turned a documented no-op into an uncaught `ValueError`. + """ + transformation = _for_each( + field_path=["a", segment], transformations=[_add_fields(["added"], "value")] + ) + record = {"a": [{"id": 0}, {"id": 1}, {"id": 2}]} + expected = copy.deepcopy(record) + + transformation.transform(record) + + assert record == expected + + +@pytest.mark.parametrize( + "segment", + [ + "-4", + "-3", + "-1", + "-0", + "0", + "2", + "3", + "+1", + " 1", + "1_0", + "\u00b2", + "\u2155", + "abc", + "", + ], +) +def test_the_plain_walk_resolves_a_list_segment_exactly_like_dpath(segment): + record = {"a": [{"id": index} for index in range(3)]} + + assert ForEach._resolve_collections(record, ["a", segment]) == list( + dpath.values(copy.deepcopy(record), ["a", segment]) + ) + + +def test_a_literal_empty_field_path_segment_is_a_manifest_error(): + """ + An empty segment cannot be the render of a template, so it is the manifest's fault, not the + config's -- the same call `_elements_of` makes for a mis-pointed `field_path`. + """ + with pytest.raises(AirbyteTracedException) as exc_info: + _for_each(field_path=["items", ""], transformations=[_add_fields(["added"], "value")]) + + assert exc_info.value.failure_type == FailureType.system_error + + +def test_a_numeric_config_value_is_a_valid_list_index_rather_than_a_config_error(): + transformation = _for_each( + field_path=["a", "{{ config['index'] }}"], + transformations=[_add_fields(["added"], "value")], + config={"index": -1}, + ) + record = {"a": [{"id": 0}, {"id": 1}]} + + transformation.transform(record) + + assert record == {"a": [{"id": 0}, {"id": 1, "added": "value"}]} + + def test_a_path_without_a_glob_never_goes_through_dpath(): """ `dpath` resolves by folding over the whole object graph, so it costs O(total nodes in the record) From 282023fd2154c350aaa1f312ccae6122db23a7d9 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Tue, 15 Sep 2026 22:58:17 +0300 Subject: [PATCH 4/4] fix: make ForEach glob handling consistent and correct the docstring claims Addresses the review findings on the ForEach transformation: - Overlapping glob matches no longer transform the same object twice. `dpath.values(record, ["**"])` yields every container at every depth, so a nested object came back both on its own and as a member of its parent list, and a non-idempotent nested transformation ran on it twice. Matches are now de-duplicated by identity. - The plain walk now matches an integer dict key from a string segment, the way `dpath` does (it glob-matches against `str(key)`). Without this, adding a glob elsewhere in the path changed which values the same `field_path` resolved. - Documented that globs are lenient: a glob selects rather than addresses, so a match that is neither an object nor a list of objects is skipped instead of raising. Raising there would make `*` unusable, since it routinely lands on unrelated scalars. - Documented that a key literally containing `*`, `?` or `[` cannot be addressed. This is inherited from `dpath`/`fnmatch`, which offer no escape syntax, and is identical to `DpathExtractor`. - Narrowed the "leaves the whole record untouched" guarantee to element validation. Nothing in the transformation pipeline is atomic, and a nested transformation that raises part-way through leaves earlier elements mutated. - Dropped the claim that the Python transformations this component replaces skip `null` elements. `MondayTransformation` raises `AttributeError` on one. Skipping is still the right behavior, just not for that reason. Tests cover each of the above, including a parity assertion against `dpath.values` for the integer dict key. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 2 +- .../models/declarative_component_schema.py | 2 +- .../declarative/transformations/for_each.py | 72 +++++++++--- .../transformations/test_for_each.py | 106 ++++++++++++++++++ 4 files changed, 162 insertions(+), 20 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 142dab2b5a..46456cd547 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -3020,7 +3020,7 @@ definitions: enum: [ForEach] field_path: title: Field Path - description: Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. Glob segments (`*`, `?`, `[...]`) are supported, in which case every matching value is iterated over. A `null` element inside the collection is skipped, because a null in an array is ordinary payload; any other non-object element raises an error, because it cannot be transformed in place. + description: Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. A `null` element inside the collection is skipped, because a null in an array is ordinary payload; any other non-object element raises an error, because it cannot be transformed in place. Glob segments (`*`, `?`, `[...]`) are supported, in which case every matching value is iterated over; because a glob selects rather than addresses a value, a match that is neither an object nor a list of objects is skipped instead of raising, and a value that the glob matches more than once (as `**` does) is still transformed only once. type: array items: type: string diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index cc040bf637..be2141a604 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -2335,7 +2335,7 @@ class ForEach(BaseModel): type: Literal["ForEach"] field_path: List[str] = Field( ..., - description="Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. Glob segments (`*`, `?`, `[...]`) are supported, in which case every matching value is iterated over. A `null` element inside the collection is skipped, because a null in an array is ordinary payload; any other non-object element raises an error, because it cannot be transformed in place.", + description="Path to the collection whose elements the nested transformations are applied to. If the path does not resolve, or resolves to a scalar or null, the transformation is a no-op. If it resolves to an object, that object is treated as a collection of one. A `null` element inside the collection is skipped, because a null in an array is ordinary payload; any other non-object element raises an error, because it cannot be transformed in place. Glob segments (`*`, `?`, `[...]`) are supported, in which case every matching value is iterated over; because a glob selects rather than addresses a value, a match that is neither an object nor a list of objects is skipped instead of raising, and a value that the glob matches more than once (as `**` does) is still transformed only once.", examples=[ ["column_values"], ["data", "assets"], diff --git a/airbyte_cdk/sources/declarative/transformations/for_each.py b/airbyte_cdk/sources/declarative/transformations/for_each.py index 8d41b3d05e..82370b983b 100644 --- a/airbyte_cdk/sources/declarative/transformations/for_each.py +++ b/airbyte_cdk/sources/declarative/transformations/for_each.py @@ -3,7 +3,7 @@ # from dataclasses import InitVar, dataclass -from typing import Any, Dict, List, Mapping, Optional, Union +from typing import Any, Dict, List, Mapping, Optional, Set, Union import dpath @@ -19,17 +19,27 @@ _GLOB_CHARACTERS = ("*", "?", "[") -def _as_list_index(segment: str, length: int) -> Optional[int]: +def _as_list_index_or_none(segment: str) -> Optional[int]: """ - Resolve `segment` against a list of `length` elements exactly the way `dpath` does: `int()` is - what decides whether a segment is an index at all (`dpath.segments.match`), and a negative index - counts from the end (`dpath.types.ListIndex.__eq__`). A segment `int()` rejects, or an index - outside the list, matches nothing -- the empty result `dpath.values` returns. + `int()` is what decides whether a segment is an index at all (`dpath.segments.match`), so a + segment `int()` rejects is not an index. Note that `int()` is not `str.isdigit()`: `"²".isdigit()` + is `True` while `int("²")` raises. """ try: - index = int(segment) + return int(segment) except ValueError: return None + + +def _as_list_index(segment: str, length: int) -> Optional[int]: + """ + Resolve `segment` against a list of `length` elements exactly the way `dpath` does: a negative + index counts from the end (`dpath.types.ListIndex.__eq__`). A segment that is not an index, or an + index outside the list, matches nothing -- the empty result `dpath.values` returns. + """ + index = _as_list_index_or_none(segment) + if index is None: + return None return index if -length <= index < length else None @@ -51,7 +61,14 @@ class ForEach(RecordTransformation): * A segment addressing a list is resolved the way `dpath` resolves it: any value `int()` accepts is an index, and a negative index counts from the end. An index outside the list matches nothing. * Glob segments (`*`, `?`, `[...]`) are supported. When the path contains one, every value matching the - glob is treated as its own collection to iterate over. + glob is treated as its own collection to iterate over. Because a glob selects rather than addresses, + it is lenient: a match that is neither an object nor a list of objects is skipped instead of raising. + A glob can also match the same object twice -- `**` matches a nested object both on its own and as a + member of its parent list -- so matches are de-duplicated by identity and every object is transformed + exactly once. + * A key that literally contains `*`, `?` or `[` cannot be addressed: the segment is matched as a glob, + which is how `dpath` and therefore every other dpath-based component (`DpathExtractor`, + `DpathFlattenFields`) already behaves. There is no escape syntax. * If the path does not resolve, the transformation is a no-op. A collection missing from some records is expected and must not fail the sync. * If the resolved value is a list, the nested transformations are applied to each element. @@ -60,13 +77,16 @@ class ForEach(RecordTransformation): * If the resolved value is a scalar or `None`, the transformation is a no-op. * An empty `field_path` resolves to the record itself, the same way `DpathExtractor` treats an empty `field_path`. - * A `null` element inside the collection is skipped. A `null` in an array is ordinary API payload, not a - manifest mistake, and the Python transformations this component replaces skip it too. - * Any other non-object element (a string, a number) raises. Such an element cannot be mutated in place, - and silently skipping it would hide a mis-pointed `field_path` behind missing data. + * A `null` element inside the collection is skipped. A `null` in an array is ordinary API payload rather + than a manifest mistake, so it must not fail the sync. + * Any other non-object element (a string, a number) in a list addressed without a glob raises. Such an + element cannot be mutated in place, and silently skipping it would hide a mis-pointed `field_path` + behind missing data. - Every collection the path matches is resolved and validated before any element is transformed, so a bad - element anywhere under a glob leaves the whole record untouched. + Every collection the path matches is resolved and validated before any element is transformed, so a + non-object element anywhere under a glob leaves the whole record untouched. This covers element + validation only: like every other transformation in the pipeline, a nested transformation that raises + part-way through leaves the elements already transformed mutated. Example: ```yaml @@ -124,12 +144,22 @@ def transform( # Resolving and validating every matched collection up front is what makes the documented # "a bad element leaves the record untouched" contract hold across a glob, not just within a # single collection. - elements = [ + candidates = [ element for collection in self._resolve_collections(record, path) for element in self._elements_of(collection, path) ] + # Overlapping glob matches can yield the same object twice -- `**` matches a nested object + # both on its own and as a member of its parent list. De-duplicate by identity so a + # non-idempotent nested transformation is not applied twice to one object. + seen: Set[int] = set() + elements = [] + for element in candidates: + if id(element) not in seen: + seen.add(id(element)) + elements.append(element) + for element in elements: for transformation in self.transformations: transformation.transform( @@ -177,9 +207,15 @@ def _resolve_collections(record: Dict[str, Any], path: List[str]) -> List[Any]: node: Any = record for segment in path: if isinstance(node, dict): - if segment not in node: - return [] - node = node[segment] + if segment in node: + node = node[segment] + else: + # `dpath` glob-matches a segment against `str(key)`, so a string segment + # addresses an integer dict key too. Keep the walk in parity with it. + index = _as_list_index_or_none(segment) + if index is None or index not in node: + return [] + node = node[index] elif isinstance(node, list): index = _as_list_index(segment, len(node)) if index is None: diff --git a/unit_tests/sources/declarative/transformations/test_for_each.py b/unit_tests/sources/declarative/transformations/test_for_each.py index 9403f289a9..f091247393 100644 --- a/unit_tests/sources/declarative/transformations/test_for_each.py +++ b/unit_tests/sources/declarative/transformations/test_for_each.py @@ -601,6 +601,112 @@ def test_the_plain_walk_resolves_a_list_segment_exactly_like_dpath(segment): ) +class _CountingTransformation(RecordTransformation): + """Counts how many times each object it is handed has been transformed.""" + + def transform(self, record, config=None, stream_state=None, stream_slice=None) -> None: + record["applied"] = record.get("applied", 0) + 1 + + +def test_a_glob_match_that_is_not_an_object_is_skipped_instead_of_raising(): + """ + A glob selects rather than addresses, so it routinely lands on scalars that have nothing to do + with the intended collection. Raising on those would make `*` unusable, so globs are lenient -- + unlike `["items"]`, which addresses the list itself and does raise on a scalar element. + """ + transformation = _for_each( + field_path=["items", "*"], transformations=[_add_fields(["added"], "value")] + ) + record = {"items": ["a", "b"]} + + transformation.transform(record) + + assert record == {"items": ["a", "b"]} + + with pytest.raises(AirbyteTracedException): + _for_each( + field_path=["items"], transformations=[_add_fields(["added"], "value")] + ).transform({"items": ["a", "b"]}) + + +def test_overlapping_glob_matches_transform_each_object_exactly_once(): + """ + `dpath.values(record, ["**"])` yields every container at every depth, so a nested object comes + back both on its own and as a member of its parent list. Without the identity de-duplication a + non-idempotent nested transformation would be applied twice to the same object. + """ + transformation = _for_each(field_path=["**"], transformations=[_CountingTransformation()]) + record = {"a": [{"id": 1, "b": [{"id": 2}]}]} + + transformation.transform(record) + + assert record == {"a": [{"id": 1, "applied": 1, "b": [{"id": 2, "applied": 1}]}]} + + +def test_the_plain_walk_matches_an_integer_dict_key_like_dpath(): + """ + `dpath` glob-matches a segment against `str(key)`, so a string segment addresses an integer dict + key too. The plain walk has to agree, or adding a glob elsewhere in the path would change which + values the same manifest resolves. + """ + record = {"a": {1: [{"id": 1}]}} + + assert ForEach._resolve_collections(record, ["a", "1"]) == list( + dpath.values(copy.deepcopy(record), ["a", "1"]) + ) + + transformation = _for_each( + field_path=["a", "1"], transformations=[_add_fields(["added"], "value")] + ) + transformation.transform(record) + + assert record == {"a": {1: [{"id": 1, "added": "value"}]}} + + +def test_a_key_that_literally_contains_a_glob_character_cannot_be_addressed(): + """ + Documented limitation, inherited from `dpath` itself rather than introduced here: the segment is + glob-matched, and neither `dpath` nor `fnmatch` offers an escape syntax. `DpathExtractor` and + `DpathFlattenFields` behave identically. + """ + record = {"a[0]": [{"id": 1}]} + + assert list(dpath.values(copy.deepcopy(record), ["a[0]"])) == [] + + transformation = _for_each( + field_path=["a[0]"], transformations=[_add_fields(["added"], "value")] + ) + transformation.transform(record) + + assert record == {"a[0]": [{"id": 1}]} + + +def test_a_nested_transformation_that_raises_leaves_the_earlier_elements_mutated(): + """ + The up-front validation covers element *shape* only. Nothing in the transformation pipeline is + atomic, and `ForEach` does not pretend otherwise. + """ + + class _RaisesOnTheSecondElement(RecordTransformation): + def __init__(self): + self.calls = 0 + + def transform(self, record, config=None, stream_state=None, stream_slice=None) -> None: + self.calls += 1 + if self.calls == 2: + raise RuntimeError("boom") + record["added"] = "value" + + record = {"items": [{"id": 1}, {"id": 2}]} + + with pytest.raises(RuntimeError): + _for_each(field_path=["items"], transformations=[_RaisesOnTheSecondElement()]).transform( + record + ) + + assert record == {"items": [{"id": 1, "added": "value"}, {"id": 2}]} + + def test_a_literal_empty_field_path_segment_is_a_manifest_error(): """ An empty segment cannot be the render of a template, so it is the manifest's fault, not the