feat: add NestedRecordExtractor for extracting nested collections with parent context

airbyte_cdk/__init__.py

@@ -103,7 +103,12 @@
 from .sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from .sources.declarative.decoders import Decoder, JsonDecoder
 from .sources.declarative.exceptions import ReadException
-from .sources.declarative.extractors import DpathExtractor, RecordSelector
+from .sources.declarative.extractors import (
+    DpathExtractor,
+    NestedRecordExtractor,
+    ParentFieldPath,
+    RecordSelector,
+)
 from .sources.declarative.extractors.record_extractor import RecordExtractor
 from .sources.declarative.extractors.record_filter import RecordFilter
 from .sources.declarative.interpolation import InterpolatedBoolean, InterpolatedString
@@ -254,10 +259,12 @@
     "JsonFileSchemaLoader",
     "LegacyToPerPartitionStateMigration",
     "MinMaxDatetime",
+    "NestedRecordExtractor",
     "NoAuth",
     "OffsetIncrement",
     "PageIncrement",
     "PaginationStrategy",
+    "ParentFieldPath",
     "ParentStreamConfig",
     "ReadException",
     "RecordExtractor",

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2203,6 +2203,88 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  NestedRecordExtractor:
+    title: Nested Record Extractor
+    description: Record extractor that yields the records of a collection nested inside another record, while carrying fields down from the record that contains it. For every record the parent extractor yields, the nested collection is read from the child field path and each of its elements is emitted, with the configured parent fields copied onto it first.
+    type: object
+    required:
+      - type
+      - parent_extractor
+      - child_field_path
+    properties:
+      type:
+        type: string
+        enum: [NestedRecordExtractor]
+      parent_extractor:
+        title: Parent Extractor
+        description: Extractor producing the records that contain the nested collection. Nest another NestedRecordExtractor here to reach a collection more than one level down, or to copy fields from more than one ancestor. The type is required here, because this field accepts a NestedRecordExtractor as readily as a DpathExtractor.
+        anyOf:
+          - "$ref": "#/definitions/DpathExtractor"
+          - "$ref": "#/definitions/NestedRecordExtractor"
+          - "$ref": "#/definitions/CustomRecordExtractor"
+      child_field_path:
+        title: Child Field Path
+        description: List of potentially nested fields describing the full path of the nested collection on each parent record. The collection may be a list or a single object. A path that does not resolve yields no records. The "*" wildcard is not supported and is rejected rather than read as a field name; flatten across several collections in the parent extractor instead. Every other character addresses the field that spells it.
+        type: array
+        minItems: 1
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["reviews", "nodes"]
+          - ["comments"]
+          - ["{{ parameters.child_connection }}", "nodes"]
+      parent_fields:
+        title: Parent Fields
+        description: Fields to copy from the parent record onto every record extracted from the nested collection. A parent path that is absent on the parent, or that runs into a null, copies null; one that runs into a value holding no such field fails the sync rather than copying null silently. A key already present on the child record is overwritten.
+        type: array
+        items:
+          "$ref": "#/definitions/ParentFieldPath"
+        examples:
+          - - parent_path: ["url"]
+              record_path: ["pull_request_url"]
+          - - parent_path: ["id"]
+              record_path: ["pull_request_id"]
+            - parent_path: ["author", "login"]
+              record_path: ["pull_request_author"]
+      $parameters:
+        type: object
+        additionalProperties: true
+  ParentFieldPath:
+    title: Parent Field Path
+    description: A single copy instruction, reading a field from the parent record and writing it onto every record extracted from the nested collection.
+    type: object
+    required:
+      - parent_path
+      - record_path
+    properties:
+      parent_path:
+        title: Parent Path
+        description: List of potentially nested fields describing the full path of the field to read on the parent record. The "*" wildcard is not supported and is rejected rather than read as a field name.
+        type: array
+        minItems: 1
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["url"]
+          - ["id"]
+          - ["author", "login"]
+      record_path:
+        title: Record Path
+        description: List of potentially nested fields describing where to write the value on the extracted record. Intermediate objects are created as needed, and an intermediate that is present but is not an object is replaced by one. Every segment is an object key, so a numeric segment does not create a list.
+        type: array
+        minItems: 1
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["pull_request_url"]
+          - ["comment_id"]
+          - ["parent", "id"]
   ResponseToFileExtractor:
     title: CSV To File Extractor
     description: A record extractor designed for handling large responses that may exceed memory limits (to prevent OOM issues). It downloads a CSV file to disk, reads the data from disk, and deletes the file once it has been fully processed.
@@ -3948,6 +4030,7 @@ definitions:
       extractor:
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
+          - "$ref": "#/definitions/NestedRecordExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
       record_filter:
         title: Record Filter

airbyte_cdk/sources/declarative/extractors/__init__.py

@@ -4,6 +4,10 @@
 
 from airbyte_cdk.sources.declarative.extractors.dpath_extractor import DpathExtractor
 from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
+from airbyte_cdk.sources.declarative.extractors.nested_record_extractor import (
+    NestedRecordExtractor,
+    ParentFieldPath,
+)
 from airbyte_cdk.sources.declarative.extractors.record_filter import RecordFilter
 from airbyte_cdk.sources.declarative.extractors.record_selector import RecordSelector
 from airbyte_cdk.sources.declarative.extractors.response_to_file_extractor import (
@@ -15,6 +19,8 @@
     "TypeTransformer",
     "HttpSelector",
     "DpathExtractor",
+    "NestedRecordExtractor",
+    "ParentFieldPath",
     "RecordFilter",
     "RecordSelector",
     "ResponseToFileExtractor",