feat: Add expand_records_from_field and remain_original_record to DpathExtractor

- Add optional expand_records_from_field parameter to extract items from nested arrays
- Add optional remain_original_record parameter to preserve parent record context
- Implement _expand_record method to handle array expansion logic
- Add comprehensive unit tests covering all edge cases
- Maintain backward compatibility with existing functionality

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/extractors/dpath_extractor.py

@@ -3,7 +3,7 @@
 #
 
 from dataclasses import InitVar, dataclass, field
-from typing import Any, Iterable, List, Mapping, MutableMapping, Union
+from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Union
 
 import dpath
 import requests
@@ -24,6 +24,12 @@ class DpathExtractor(RecordExtractor):
     If the field path points to an empty object, an empty array is returned.
     If the field path points to a non-existing path, an empty array is returned.
 
+    Optionally, records can be expanded by extracting items from a nested array field.
+    When expand_records_from_field is configured, each extracted record is expanded by
+    extracting items from the specified nested array path and emitting each item as a
+    separate record. If remain_original_record is True, each expanded record will include
+    the original parent record in an "original_record" field.
+
     Examples of instantiating this transform:
     ```
       extractor:
@@ -47,16 +53,32 @@ class DpathExtractor(RecordExtractor):
         field_path: []
     ```
 
+    ```
+      extractor:
+        type: DpathExtractor
+        field_path:
+          - "data"
+          - "object"
+        expand_records_from_field:
+          - "lines"
+          - "data"
+        remain_original_record: true
+    ```
+
     Attributes:
         field_path (Union[InterpolatedString, str]): Path to the field that should be extracted
         config (Config): The user-provided configuration as specified by the source's spec
         decoder (Decoder): The decoder responsible to transfom the response in a Mapping
+        expand_records_from_field (Optional[List[Union[InterpolatedString, str]]]): Path to a nested array field within each extracted record. If provided, items from this array will be extracted and emitted as separate records.
+        remain_original_record (bool): If True and expand_records_from_field is set, each expanded record will include the original parent record in an "original_record" field. Defaults to False.
     """
 
     field_path: List[Union[InterpolatedString, str]]
     config: Config
     parameters: InitVar[Mapping[str, Any]]
     decoder: Decoder = field(default_factory=lambda: JsonDecoder(parameters={}))
+    expand_records_from_field: Optional[List[Union[InterpolatedString, str]]] = None
+    remain_original_record: bool = False
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._field_path = [
@@ -67,6 +89,44 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
                 self._field_path[path_index] = InterpolatedString.create(
                     self.field_path[path_index], parameters=parameters
                 )
+        
+        if self.expand_records_from_field:
+            self._expand_path = [
+                InterpolatedString.create(path, parameters=parameters) 
+                for path in self.expand_records_from_field
+            ]
+        else:
+            self._expand_path = None
+
+    def _expand_record(self, record: MutableMapping[Any, Any]) -> Iterable[MutableMapping[Any, Any]]:
+        """Expand a record by extracting items from a nested array field."""
+        if not self._expand_path:
+            yield record
+            return
+        
+        expand_path = [path.eval(self.config) for path in self._expand_path]
+        
+        try:
+            nested_array = dpath.get(record, expand_path)
+        except (KeyError, TypeError):
+            yield record
+            return
+        
+        if not isinstance(nested_array, list):
+            yield record
+            return
+        
+        if len(nested_array) == 0:
+            return
+        
+        for item in nested_array:
+            if isinstance(item, dict):
+                expanded_record = dict(item)
+                if self.remain_original_record:
+                    expanded_record["original_record"] = record
+                yield expanded_record
+            else:
+                yield item
 
     def extract_records(self, response: requests.Response) -> Iterable[MutableMapping[Any, Any]]:
         for body in self.decoder.decode(response):
@@ -79,8 +139,9 @@ def extract_records(self, response: requests.Response) -> Iterable[MutableMappin
                 else:
                     extracted = dpath.get(body, path, default=[])  # type: ignore # extracted will be a MutableMapping, given input data structure
             if isinstance(extracted, list):
-                yield from extracted
+                for record in extracted:
+                    yield from self._expand_record(record)
             elif extracted:
-                yield extracted
+                yield from self._expand_record(extracted)
             else:
                 yield from []

unit_tests/sources/declarative/extractors/test_dpath_extractor.py

@@ -121,3 +121,137 @@ def test_dpath_extractor(field_path: List, decoder: Decoder, body, expected_reco
     actual_records = list(extractor.extract_records(response))
 
     assert actual_records == expected_records
+
+
+@pytest.mark.parametrize(
+    "field_path, expand_records_from_field, remain_original_record, body, expected_records",
+    [
+        (
+            ["data", "object"],
+            ["lines", "data"],
+            False,
+            {
+                "data": {
+                    "object": {
+                        "id": "in_123",
+                        "created": 1234567890,
+                        "lines": {
+                            "data": [
+                                {"id": "il_1", "amount": 100},
+                                {"id": "il_2", "amount": 200},
+                            ]
+                        },
+                    }
+                }
+            },
+            [
+                {"id": "il_1", "amount": 100},
+                {"id": "il_2", "amount": 200},
+            ],
+        ),
+        (
+            ["data", "object"],
+            ["lines", "data"],
+            True,
+            {
+                "data": {
+                    "object": {
+                        "id": "in_123",
+                        "created": 1234567890,
+                        "lines": {
+                            "data": [
+                                {"id": "il_1", "amount": 100},
+                            ]
+                        },
+                    }
+                }
+            },
+            [
+                {
+                    "id": "il_1",
+                    "amount": 100,
+                    "original_record": {
+                        "id": "in_123",
+                        "created": 1234567890,
+                        "lines": {"data": [{"id": "il_1", "amount": 100}]},
+                    },
+                },
+            ],
+        ),
+        (
+            ["data"],
+            ["items"],
+            False,
+            {"data": {"id": "parent_1", "items": []}},
+            [],
+        ),
+        (
+            ["data"],
+            ["items"],
+            False,
+            {"data": {"id": "parent_1"}},
+            [{"id": "parent_1"}],
+        ),
+        (
+            ["data"],
+            ["items"],
+            False,
+            {"data": {"id": "parent_1", "items": "not_an_array"}},
+            [{"id": "parent_1", "items": "not_an_array"}],
+        ),
+        (
+            ["data"],
+            ["nested", "array"],
+            False,
+            {"data": {"id": "parent_1", "nested": {"array": [{"id": "child_1"}, {"id": "child_2"}]}}},
+            [{"id": "child_1"}, {"id": "child_2"}],
+        ),
+        (
+            ["data"],
+            ["items"],
+            False,
+            {"data": {"id": "parent_1", "items": [1, 2, "string", {"id": "dict_item"}]}},
+            [1, 2, "string", {"id": "dict_item"}],
+        ),
+        (
+            [],
+            ["items"],
+            False,
+            [
+                {"id": "parent_1", "items": [{"id": "child_1"}]},
+                {"id": "parent_2", "items": [{"id": "child_2"}, {"id": "child_3"}]},
+            ],
+            [{"id": "child_1"}, {"id": "child_2"}, {"id": "child_3"}],
+        ),
+    ],
+    ids=[
+        "test_expand_nested_array",
+        "test_expand_with_original_record",
+        "test_expand_empty_array_yields_nothing",
+        "test_expand_missing_path_yields_original",
+        "test_expand_non_array_yields_original",
+        "test_expand_deeply_nested_path",
+        "test_expand_mixed_types_in_array",
+        "test_expand_multiple_parent_records",
+    ],
+)
+def test_dpath_extractor_with_expansion(
+    field_path: List,
+    expand_records_from_field: List,
+    remain_original_record: bool,
+    body,
+    expected_records: List,
+):
+    extractor = DpathExtractor(
+        field_path=field_path,
+        config=config,
+        decoder=decoder_json,
+        parameters=parameters,
+        expand_records_from_field=expand_records_from_field,
+        remain_original_record=remain_original_record,
+    )
+
+    response = create_response(body)
+    actual_records = list(extractor.extract_records(response))
+
+    assert actual_records == expected_records