[DO NOT MERGE] prerelease: CombinedExtractor (#1162) + RecordExpander parent_fields (#1165)

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 (
+    CombinedExtractor,
+    CombineMode,
+    DpathExtractor,
+    RecordSelector,
+)
 from .sources.declarative.extractors.record_extractor import RecordExtractor
 from .sources.declarative.extractors.record_filter import RecordFilter
 from .sources.declarative.interpolation import InterpolatedBoolean, InterpolatedString
@@ -234,6 +239,8 @@
     "BasicHttpAuthenticator",
     "BearerAuthenticator",
     "CartesianProductStreamSlicer",
+    "CombineMode",
+    "CombinedExtractor",
     "CursorPaginationStrategy",
     "DatetimeBasedCursor",
     "DeclarativeAuthenticator",

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1917,6 +1917,7 @@ definitions:
             anyOf:
               - "$ref": "#/definitions/DpathExtractor"
               - "$ref": "#/definitions/CustomRecordExtractor"
+              - "$ref": "#/definitions/CombinedExtractor"
           file_extractor:
             description: Responsible for fetching the content of the file. If not defined, the assumption is that the whole response body is the file content
             anyOf:
@@ -2189,6 +2190,66 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  CombineMode:
+    title: Combine Mode
+    description: 'How the records of the sub-extractors of a CombinedExtractor are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record.'
+    type: string
+    enum:
+      - union
+      - first_match
+    examples:
+      - union
+      - first_match
+  CombinedExtractor:
+    title: Combined Extractor
+    description: Record extractor that combines the output of several sub-extractors. Use it when a single response holds records under more than one path, when the records can be found under one of several alternative paths, or when a record has to be assembled from several parallel lists.
+    type: object
+    required:
+      - type
+      - extractors
+    properties:
+      type:
+        type: string
+        enum: [CombinedExtractor]
+      extractors:
+        title: Extractors
+        description: The record extractors to combine. At least one is required. Each sub-extractor is given the same HTTP response and decodes it independently, so the response is parsed once per sub-extractor. Streaming decoders (CsvDecoder, JsonlDecoder, JsonItemsDecoder, GzipDecoder, IterableDecoder) can only read the response body once and are rejected.
+        type: array
+        minItems: 1
+        items:
+          anyOf:
+            - "$ref": "#/definitions/DpathExtractor"
+            - "$ref": "#/definitions/CustomRecordExtractor"
+            - "$ref": "#/definitions/CombinedExtractor"
+      mode:
+        title: Combine Mode
+        description: 'How the records of the sub-extractors are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record. Note that a paginator which counts the records of a page counts the combined records: under "union" that is the sum over all sub-extractors, which overshoots the API page size, so "union" is rejected with an OffsetIncrement paginator because the offset would skip records. Under "first_match" the count is the count of the winning sub-extractor, which is usually the number of records the API returned for the page.'
+        "$ref": "#/definitions/CombineMode"
+        default: union
+      skip_empty_records:
+        title: Skip Empty Records
+        description: 'Whether to drop empty records - null, {}, [], "" - yielded by a sub-extractor before the records are combined. Off by default, so the output of a sub-extractor is passed through as it is. Turn it on when the API can return nulls in the middle of a record list, which a GraphQL API does when it answers a partial response and reports the failure in a sibling error field. Under "first_match" this also changes which sub-extractor wins: a sub-extractor whose records are all empty no longer counts as a match, so the next one is tried. The dropped records still count toward the page size of an OffsetIncrement or PageIncrement paginator, so a full page holding a null does not end pagination early.'
+        type: boolean
+        default: false
+      $parameters:
+        type: object
+        additionalProperties: true
+    examples:
+      - type: CombinedExtractor
+        mode: union
+        extractors:
+          - type: DpathExtractor
+            field_path: ["data", "repository", "issues", "nodes"]
+          - type: DpathExtractor
+            field_path: ["data", "repository", "pullRequests", "nodes"]
+      - type: CombinedExtractor
+        mode: first_match
+        skip_empty_records: true
+        extractors:
+          - type: DpathExtractor
+            field_path: ["data", "boards", "*", "items_page", "items", "*"]
+          - type: DpathExtractor
+            field_path: ["data", "next_items_page", "items"]
   DpathExtractor:
     title: Dpath Extractor
     description: Record extractor that searches a decoded response over a path defined as an array of fields.
@@ -2267,6 +2328,17 @@ definitions:
         description: If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.
         type: boolean
         default: false
+      parent_fields:
+        title: Parent Fields
+        description: 'Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. A copied object or array is deep-copied, so a transformation that writes inside it affects only that item. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.'
+        type: array
+        items:
+          "$ref": "#/definitions/ParentFieldPath"
+      merge_parent:
+        title: Merge Parent
+        description: If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. Each item receives its own deep copy of the merged parent fields, so a transformation that writes into a nested value affects only that item. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.
+        type: boolean
+        default: false
       on_no_records:
         title: On No Records
         description: Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged.
@@ -2294,6 +2366,45 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  ParentFieldPath:
+    title: Parent Field Path
+    description: One value to copy from the record being expanded onto each expanded item. Both paths are lists, so a value nested inside the parent can be copied into a nested position on the item; intermediate objects on `record_path` are created as needed. Glob characters (`*`, `?`, `[`) are not supported in either path, since each must identify a single field; this is enforced on the interpolated values.
+    type: object
+    required:
+      - type
+      - parent_path
+      - record_path
+    properties:
+      type:
+        type: string
+        enum: [ParentFieldPath]
+      parent_path:
+        title: Parent Path
+        description: Path to the value on the record being expanded.
+        type: array
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["url"]
+          - ["id"]
+          - ["repository", "name"]
+      record_path:
+        title: Record Path
+        description: Path on the expanded item to write the value to. An existing value is overwritten. Intermediate objects are created as needed, so a path may not pass through a value the item already holds as a scalar, and a numeric path segment creates an array rather than an object - the same behavior as `AddFields`.
+        type: array
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["pull_request_url"]
+          - ["comment_id"]
+          - ["parent", "id"]
+      $parameters:
+        type: object
+        additionalProperties: true
   ExponentialBackoffStrategy:
     title: Exponential Backoff
     description: Backoff strategy with an exponential backoff interval. The interval is defined as factor * 2^attempt_count.
@@ -3984,6 +4095,7 @@ definitions:
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       record_filter:
         title: Record Filter
         description: Responsible for filtering records to be emitted by the Source.
@@ -4546,17 +4658,20 @@ definitions:
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       download_target_extractor:
         description: Responsible for fetching the final result `urls` provided by the completed / finished / ready async job.
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       download_extractor:
         description: Responsible for fetching the records from provided urls.
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
           - "$ref": "#/definitions/ResponseToFileExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       creation_requester:
         description: Requester component that describes how to prepare HTTP requests to send to the source API to create the async server-side job.
         anyOf:

airbyte_cdk/sources/declarative/expanders/__init__.py

@@ -2,6 +2,10 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from airbyte_cdk.sources.declarative.expanders.record_expander import OnNoRecords, RecordExpander
+from airbyte_cdk.sources.declarative.expanders.record_expander import (
+    OnNoRecords,
+    ParentFieldPath,
+    RecordExpander,
+)
 
-__all__ = ["OnNoRecords", "RecordExpander"]
+__all__ = ["OnNoRecords", "ParentFieldPath", "RecordExpander"]