diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index aa3bbbc5e0..46456cd547 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,59 @@ 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 `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 + interpolation_context: + - config + - parameters + 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..be2141a604 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 `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"], + ["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/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/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..82370b983b --- /dev/null +++ b/airbyte_cdk/sources/declarative/transformations/for_each.py @@ -0,0 +1,253 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +from dataclasses import InitVar, dataclass +from typing import Any, Dict, List, Mapping, Optional, Set, Union + +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 + +# 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 = ("*", "?", "[") + + +def _as_list_index_or_none(segment: str) -> Optional[int]: + """ + `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: + 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 + + +@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 segment that + 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. 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. + * 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`. + * 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 + 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 + 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._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, + 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 = 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. + 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( + 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) + # 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=( + 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. 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 any(character in segment for segment in path for character in _GLOB_CHARACTERS): + return list(dpath.values(record, path)) + + node: Any = record + for segment in path: + if isinstance(node, dict): + 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: + return [] + node = node[index] + 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 isinstance(element, dict): + elements.append(element) + elif element is not None: + raise AirbyteTracedException( + message=( + 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.system_error, + ) + 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 21c99adc71..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 @@ -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, @@ -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, ) @@ -3407,6 +3408,116 @@ 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"}]}]} + + 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 new file mode 100644 index 0000000000..f091247393 --- /dev/null +++ b/unit_tests/sources/declarative/transformations/test_for_each.py @@ -0,0 +1,839 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +import copy +import time +from typing import Any, Mapping +from unittest.mock import patch + +import dpath +import pytest +from jsonschema import ValidationError, validate + +from airbyte_cdk.models import FailureType +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, +) +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 = copy.deepcopy(record) + + 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_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"}, + ) + + with pytest.raises(AirbyteTracedException) as exc_info: + transformation.transform({"tags": ["a"]}) + + assert "the_stream" not in exc_info.value.message + assert "stream" not in exc_info.value.message + + +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): + transformation.transform(record) + + 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(): + 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"}]} + + +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": 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): + """ + 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 + ) + + +@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]) + ) + + +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 + 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) + 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"))