refactor: Extract record expansion logic into RecordExpander class

- Create new RecordExpander class in airbyte_cdk/sources/declarative/expanders/
- Move expand_records_from_field and remain_original_record parameters from DpathExtractor to RecordExpander
- Update DpathExtractor to accept optional record_expander attribute
- Register RecordExpander in manifest component transformer
- Update unit tests to use new RecordExpander class structure
- All 24 tests passing, MyPy and Ruff checks passing

This refactoring improves separation of concerns by isolating record expansion logic into a dedicated component.

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/expanders/__init__.py

@@ -0,0 +1,7 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from airbyte_cdk.sources.declarative.expanders.record_expander import RecordExpander
+
+__all__ = ["RecordExpander"]

airbyte_cdk/sources/declarative/expanders/record_expander.py

@@ -0,0 +1,78 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import InitVar, dataclass
+from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Union
+
+import dpath
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.sources.types import Config
+
+
+@dataclass
+class RecordExpander:
+    """
+    Expands records by extracting items from a nested array field.
+
+    When configured, this component extracts items from a specified nested array path
+    within each record and emits each item as a separate record. Optionally, the original
+    parent record can be embedded in each expanded item for context preservation.
+
+    Examples of instantiating this component:
+    ```
+      record_expander:
+        type: RecordExpander
+        expand_records_from_field:
+          - "lines"
+          - "data"
+        remain_original_record: true
+    ```
+
+    Attributes:
+        expand_records_from_field (List[Union[InterpolatedString, str]]): Path to a nested array field within each record. Items from this array will be extracted and emitted as separate records.
+        remain_original_record (bool): If True, each expanded record will include the original parent record in an "original_record" field. Defaults to False.
+        config (Config): The user-provided configuration as specified by the source's spec
+    """
+
+    expand_records_from_field: List[Union[InterpolatedString, str]]
+    config: Config
+    parameters: InitVar[Mapping[str, Any]]
+    remain_original_record: bool = False
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self._expand_path: Optional[List[InterpolatedString]] = [
+            InterpolatedString.create(path, parameters=parameters)
+            for path in self.expand_records_from_field
+        ]
+
+    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

airbyte_cdk/sources/declarative/extractors/dpath_extractor.py

@@ -9,6 +9,7 @@
 import requests
 
 from airbyte_cdk.sources.declarative.decoders import Decoder, JsonDecoder
+from airbyte_cdk.sources.declarative.expanders.record_expander import RecordExpander
 from airbyte_cdk.sources.declarative.extractors.record_extractor import RecordExtractor
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.types import Config
@@ -24,11 +25,10 @@ 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.
+    Optionally, records can be expanded by providing a RecordExpander component.
+    When record_expander is configured, each extracted record is passed through the
+    expander which extracts items from nested array fields and emits each item as a
+    separate record.
 
     Examples of instantiating this transform:
     ```
@@ -59,26 +59,26 @@ class DpathExtractor(RecordExtractor):
         field_path:
           - "data"
           - "object"
-        expand_records_from_field:
-          - "lines"
-          - "data"
-        remain_original_record: true
+        record_expander:
+          type: RecordExpander
+          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.
+        record_expander (Optional[RecordExpander]): Optional component to expand records by extracting items from nested array fields
     """
 
     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
+    record_expander: Optional[RecordExpander] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._field_path = [
@@ -90,46 +90,6 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
                     self.field_path[path_index], parameters=parameters
                 )
 
-        if self.expand_records_from_field:
-            self._expand_path: Optional[List[InterpolatedString]] = [
-                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):
             if len(self._field_path) == 0:
@@ -142,8 +102,14 @@ def extract_records(self, response: requests.Response) -> Iterable[MutableMappin
                     extracted = dpath.get(body, path, default=[])  # type: ignore # extracted will be a MutableMapping, given input data structure
             if isinstance(extracted, list):
                 for record in extracted:
-                    yield from self._expand_record(record)
+                    if self.record_expander:
+                        yield from self.record_expander.expand_record(record)
+                    else:
+                        yield record
             elif extracted:
-                yield from self._expand_record(extracted)
+                if self.record_expander:
+                    yield from self.record_expander.expand_record(extracted)
+                else:
+                    yield extracted
             else:
                 yield from []

airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py

@@ -42,6 +42,7 @@
     "DefaultPaginator.page_size_option": "RequestOption",
     # DpathExtractor
     "DpathExtractor.decoder": "JsonDecoder",
+    "DpathExtractor.record_expander": "RecordExpander",
     # HttpRequester
     "HttpRequester.error_handler": "DefaultErrorHandler",
     # ListPartitionRouter

unit_tests/sources/declarative/extractors/test_dpath_extractor.py

@@ -14,6 +14,7 @@
     IterableDecoder,
     JsonDecoder,
 )
+from airbyte_cdk.sources.declarative.expanders.record_expander import RecordExpander
 from airbyte_cdk.sources.declarative.extractors.dpath_extractor import DpathExtractor
 
 config = {"field": "record_array"}
@@ -247,13 +248,18 @@ def test_dpath_extractor_with_expansion(
     body,
     expected_records: List,
 ):
+    record_expander = RecordExpander(
+        expand_records_from_field=expand_records_from_field,
+        config=config,
+        parameters=parameters,
+        remain_original_record=remain_original_record,
+    )
     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,
+        record_expander=record_expander,
     )
 
     response = create_response(body)