feat: add ForEach transformation for transforming nested collections

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

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()

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",

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:

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"]