Add Combined Extractor

Author: lazebnyi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1741,6 +1741,54 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  KeyValueExtractor:
+    title: Key Value Extractor
+    description: Record extractor that extract with .
+    type: object
+    required:
+      - type
+      - keys_extractor
+      - values_extractor
+    properties:
+      type:
+        type: string
+        enum: [ KeyValueExtractor ]
+      keys_extractor:
+        description: placeholder
+        anyOf:
+          - "$ref": "#/definitions/DpathExtractor"
+          - "$ref": "#/definitions/CustomRecordExtractor"
+      values_extractor:
+        description: placeholder
+        anyOf:
+          - "$ref": "#/definitions/DpathExtractor"
+          - "$ref": "#/definitions/CustomRecordExtractor"
+      $parameters:
+        type: object
+        additionalProperties: true
+  CombinedExtractor:
+    title: Combined Extractor
+    description: Record extractor that extract with .
+    type: object
+    required:
+      - type
+      - extractors
+    properties:
+      type:
+        type: string
+        enum: [ CombinedExtractor ]
+      extractors:
+        description: placeholder
+        type: array
+        items:
+          anyOf:
+            - "$ref": "#/definitions/DpathExtractor"
+            - "$ref": "#/definitions/CombinedExtractor"
+            - "$ref": "#/definitions/KeyValueExtractor"
+            - "$ref": "#/definitions/CustomRecordExtractor"
+      $parameters:
+        type: object
+        additionalProperties: true
   DpathExtractor:
     title: Dpath Extractor
     description: Record extractor that searches a decoded response over a path defined as an array of fields.
@@ -2318,6 +2366,12 @@ definitions:
           - "$ref": "#/definitions/AsyncRetriever"
           - "$ref": "#/definitions/CustomRetriever"
           - "$ref": "#/definitions/SimpleRetriever"
+      schema_filter:
+        title: Schema Filter
+        description: placeholder
+        anyOf:
+          - "$ref": "#/definitions/RecordFilter"
+          - "$ref": "#/definitions/CustomRecordFilter"
       schema_transformations:
         title: Schema Transformations
         description: A list of transformations to be applied to the schema.
@@ -3315,6 +3369,8 @@ definitions:
       extractor:
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
+          - "$ref": "#/definitions/KeyValueExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
       record_filter:
         title: Record Filter
@@ -3994,6 +4050,9 @@ definitions:
         title: Value Type
         description: The expected data type of the value. If omitted, the type will be inferred from the value provided.
         "$ref": "#/definitions/ValueType"
+      create_or_update:
+        type: boolean
+        default: false
       $parameters:
         type: object
         additionalProperties: true
@@ -4045,6 +4104,10 @@ definitions:
           - ["data"]
           - ["data", "streams"]
           - ["data", "{{ parameters.name }}"]
+      default_values:
+        title: Default Values
+        description: placeholder
+        type: array
       $parameters:
         type: object
         additionalProperties: true
@@ -4056,7 +4119,11 @@ definitions:
         type: string
         enum: [ConfigComponentsResolver]
       stream_config:
-        "$ref": "#/definitions/StreamConfig"
+        anyOf:
+          - type: array
+            items:
+              "$ref": "#/definitions/StreamConfig"
+          - "$ref": "#/definitions/StreamConfig"
       components_mapping:
         type: array
         items:

airbyte_cdk/sources/declarative/extractors/__init__.py

@@ -3,6 +3,8 @@
 #
 
 from airbyte_cdk.sources.declarative.extractors.dpath_extractor import DpathExtractor
+from airbyte_cdk.sources.declarative.extractors.key_value_extractor import KeyValueExtractor
+from airbyte_cdk.sources.declarative.extractors.combined_extractor import CombinedExtractor
 from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
 from airbyte_cdk.sources.declarative.extractors.record_filter import RecordFilter
 from airbyte_cdk.sources.declarative.extractors.record_selector import RecordSelector
@@ -18,4 +20,6 @@
     "RecordFilter",
     "RecordSelector",
     "ResponseToFileExtractor",
+    "KeyValueExtractor",
+    "CombinedExtractor",
 ]

airbyte_cdk/sources/declarative/extractors/combined_extractor.py

@@ -0,0 +1,44 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any, Iterable, List, MutableMapping
+
+import requests
+from airbyte_cdk.sources.declarative.extractors.record_extractor import RecordExtractor
+
+
+@dataclass
+class CombinedExtractor(RecordExtractor):
+    """
+    Extractor that merges the output of multiple sub-extractors into a single record.
+
+    This extractor takes a list of `RecordExtractor` instances (`extractors`), each of which
+    independently extracts records from the response. For each response, the extractor:
+
+    1. Invokes each sub-extractor to generate iterables of records.
+    2. Zips the results together, so that the first record from each extractor is combined,
+       the second from each, and so on.
+    3. Merges each group of records into a single dictionary using `dict.update()`.
+
+    The result is a sequence of dictionaries where each dictionary contains the merged keys
+    and values from the corresponding records across all extractors.
+
+    Example:
+        keys_extractor -> yields: [{"name": "Alice", "age": 30}]
+        extra_data_extractor -> yields: [{"country": "US"}]
+        CombinedExtractor(extractors=[keys_extractor, extra_data_extractor]) ->
+            yields: [{"name": "Alice", "age": 30, "country": "US"}]
+    """
+
+    extractors: List[RecordExtractor]
+
+    def extract_records(self, response: requests.Response) -> Iterable[MutableMapping[Any, Any]]:
+        extractors_records = [extractor.extract_records(response) for extractor in self.extractors]
+
+        for records in zip(*extractors_records):
+            merged = {}
+            for record in records:
+                merged.update(record)  # merge all fields
+            yield merged

airbyte_cdk/sources/declarative/extractors/key_value_extractor.py

@@ -0,0 +1,41 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any, Iterable, MutableMapping
+
+import requests
+from itertools import islice
+from airbyte_cdk.sources.declarative.extractors.record_extractor import RecordExtractor
+
+
+@dataclass
+class KeyValueExtractor(RecordExtractor):
+    """
+    Extractor that combines keys and values from two separate extractors.
+
+    The `keys_extractor` and `values_extractor` extract records independently
+    from the response. Their outputs are zipped together to form key-value mappings.
+
+    Each key from `keys_extractor` should correspond to a key in the resulting dictionary,
+    and each value from `values_extractor` is the value for that key.
+
+    Example:
+      keys_extractor -> yields: ["name", "age"]
+      values_extractor -> yields: ["Alice", 30]
+      result: { "name": "Alice", "age": 30 }
+    """
+
+    keys_extractor: RecordExtractor
+    values_extractor: RecordExtractor
+
+    def extract_records(self, response: requests.Response) -> Iterable[MutableMapping[Any, Any]]:
+        keys = list(self.keys_extractor.extract_records(response))
+        values = self.values_extractor.extract_records(response)
+
+        while True:
+            chunk = list(islice(values, len(keys)))
+            if not chunk:
+                break
+            yield dict(zip(keys, chunk))

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -262,9 +262,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             }
         )
 
-        stream_configs = self._stream_configs(self._source_config) + self._dynamic_stream_configs(
-            self._source_config, config
-        )
+        stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
 
         api_budget_model = self._source_config.get("api_budget")
         if api_budget_model: