feat: Add on_no_records and parent_fields_to_copy to RecordExpander

devin-ai-integration[bot] · devin-ai-integration[bot] · commit 2ca9ad707cf6 · 2026-02-03T22:54:22.000Z
- Add on_no_records parameter with 'skip' (default) and 'emit_parent' options
- Add parent_fields_to_copy parameter to copy specific parent fields to child records
- Add ParentFieldMapping class to define source/target field mappings
- Update schema YAML with new properties and ParentFieldMapping definition
- Regenerate models from schema
- Add comprehensive unit tests for new features

Co-Authored-By: unknown &lt;&gt;
diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -1961,6 +1961,54 @@ 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
+      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.
+        type: string
+        enum:
+          - skip
+          - emit_parent
+        default: skip
+      parent_fields_to_copy:
+        title: Parent Fields To Copy
+        description: List of parent field mappings to copy onto each expanded child record. Each mapping specifies a source path in the parent record and a target field name in the child record.
+        type: array
+        items:
+          "$ref": "#/definitions/ParentFieldMapping"
+      $parameters:
+        type: object
+        additionalProperties: true
+  ParentFieldMapping:
+    title: Parent Field Mapping
+    description: Defines a mapping from a parent record field to a child record field.
+    type: object
+    required:
+      - type
+      - source_field_path
+      - target_field
+    properties:
+      type:
+        type: string
+        enum: [ParentFieldMapping]
+      source_field_path:
+        title: Source Field Path
+        description: Path to the field in the parent record to copy.
+        type: array
+        items:
+          type: string
+        interpolation_context:
+          - config
+        examples:
+          - ["id"]
+          - ["created"]
+          - ["metadata", "timestamp"]
+      target_field:
+        title: Target Field
+        description: Name of the field in the child record where the value will be copied.
+        type: string
+        examples:
+          - "parent_id"
+          - "subscription_updated"
       $parameters:
         type: object
         additionalProperties: true
diff --git a/airbyte_cdk/sources/declarative/expanders/__init__.py b/airbyte_cdk/sources/declarative/expanders/__init__.py
@@ -2,6 +2,9 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from airbyte_cdk.sources.declarative.expanders.record_expander import RecordExpander
+from airbyte_cdk.sources.declarative.expanders.record_expander import (
+    ParentFieldMapping,
+    RecordExpander,
+)
 
-__all__ = ["RecordExpander"]
+__all__ = ["ParentFieldMapping", "RecordExpander"]
diff --git a/airbyte_cdk/sources/declarative/expanders/record_expander.py b/airbyte_cdk/sources/declarative/expanders/record_expander.py
@@ -2,19 +2,45 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from dataclasses import InitVar, dataclass
-from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Union
+from dataclasses import InitVar, dataclass, field
+from typing import Any, Iterable, Mapping, MutableMapping
 
 import dpath
 
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.types import Config
 
 
+@dataclass
+class ParentFieldMapping:
+    """Defines a mapping from a parent record field to a child record field."""
+
+    source_field_path: list[str | InterpolatedString]
+    target_field: str
+    config: Config
+    parameters: InitVar[Mapping[str, Any]]
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self._source_path = [
+            InterpolatedString.create(path, parameters=parameters)
+            for path in self.source_field_path
+        ]
+
+    def copy_field(
+        self, parent_record: Mapping[str, Any], child_record: MutableMapping[str, Any]
+    ) -> None:
+        """Copy a field from parent record to child record."""
+        source_path = [path.eval(self.config) for path in self._source_path]
+        try:
+            value = dpath.get(dict(parent_record), source_path)
+            child_record[self.target_field] = value
+        except KeyError:
+            pass
+
+
 @dataclass
 class RecordExpander:
-    """
-    Expands records by extracting items from a nested array field.
+    """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
@@ -40,22 +66,35 @@ class RecordExpander:
           - "sections"
           - "*"
           - "items"
-        remain_original_record: false
+        on_no_records: emit_parent
+        parent_fields_to_copy:
+          - type: ParentFieldMapping
+            source_field_path: ["id"]
+            target_field: "parent_id"
     ```
 
     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. Supports wildcards (*).
-        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: Path to a nested array field within each record.
+            Items from this array will be extracted and emitted as separate records.
+            Supports wildcards (*).
+        remain_original_record: If True, each expanded record will include the original
+            parent record in an "original_record" field. Defaults to False.
+        on_no_records: Behavior when expansion produces no records. "skip" (default)
+            emits nothing. "emit_parent" emits the original parent record unchanged.
+        parent_fields_to_copy: List of field mappings to copy from parent to each
+            expanded child record.
+        config: The user-provided configuration as specified by the source's spec.
     """
 
-    expand_records_from_field: List[Union[InterpolatedString, str]]
+    expand_records_from_field: list[str | InterpolatedString]
     config: Config
     parameters: InitVar[Mapping[str, Any]]
     remain_original_record: bool = False
+    on_no_records: str = "skip"
+    parent_fields_to_copy: list[ParentFieldMapping] = field(default_factory=list)
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
-        self._expand_path: Optional[List[InterpolatedString]] = [
+        self._expand_path: list[InterpolatedString] | None = [
             InterpolatedString.create(path, parameters=parameters)
             for path in self.expand_records_from_field
         ]
@@ -68,6 +107,7 @@ def expand_record(self, record: MutableMapping[Any, Any]) -> Iterable[MutableMap
 
         parent_record = record
         expand_path = [path.eval(self.config) for path in self._expand_path]
+        expanded_any = False
 
         if "*" in expand_path:
             extracted: Any = dpath.values(parent_record, expand_path)
@@ -76,23 +116,38 @@ def expand_record(self, record: MutableMapping[Any, Any]) -> Iterable[MutableMap
                     for item in record:
                         if isinstance(item, dict):
                             expanded_record = dict(item)
-                            if self.remain_original_record:
-                                expanded_record["original_record"] = parent_record
+                            self._apply_parent_context(parent_record, expanded_record)
                             yield expanded_record
+                            expanded_any = True
                         else:
                             yield item
+                            expanded_any = True
         else:
             try:
                 extracted = dpath.get(parent_record, expand_path)
             except KeyError:
-                return
-            if not isinstance(extracted, list):
-                return
-            for item in extracted:
-                if isinstance(item, dict):
-                    expanded_record = dict(item)
-                    if self.remain_original_record:
-                        expanded_record["original_record"] = parent_record
-                    yield expanded_record
-                else:
-                    yield item
+                extracted = None
+
+            if isinstance(extracted, list):
+                for item in extracted:
+                    if isinstance(item, dict):
+                        expanded_record = dict(item)
+                        self._apply_parent_context(parent_record, expanded_record)
+                        yield expanded_record
+                        expanded_any = True
+                    else:
+                        yield item
+                        expanded_any = True
+
+        if not expanded_any and self.on_no_records == "emit_parent":
+            yield parent_record
+
+    def _apply_parent_context(
+        self, parent_record: Mapping[str, Any], child_record: MutableMapping[str, Any]
+    ) -> None:
+        """Apply parent context to a child record."""
+        if self.remain_original_record:
+            child_record["original_record"] = parent_record
+
+        for field_mapping in self.parent_fields_to_copy:
+            field_mapping.copy_field(parent_record, child_record)
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -487,23 +487,24 @@ class ResponseToFileExtractor(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
-class RecordExpander(BaseModel):
-    type: Literal["RecordExpander"]
-    expand_records_from_field: List[str] = Field(
+class OnNoRecords(Enum):
+    skip = "skip"
+    emit_parent = "emit_parent"
+
+
+class ParentFieldMapping(BaseModel):
+    type: Literal["ParentFieldMapping"]
+    source_field_path: List[str] = Field(
         ...,
-        description="Path to a nested array field within each record. Items from this array will be extracted and emitted as separate records. Supports wildcards (*) for matching multiple arrays.",
-        examples=[
-            ["lines", "data"],
-            ["items"],
-            ["nested", "array"],
-            ["sections", "*", "items"],
-        ],
-        title="Expand Records From Field",
+        description="Path to the field in the parent record to copy.",
+        examples=[["id"], ["created"], ["metadata", "timestamp"]],
+        title="Source Field Path",
     )
-    remain_original_record: Optional[bool] = Field(
-        False,
-        description='If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.',
-        title="Remain Original Record",
+    target_field: str = Field(
+        ...,
+        description="Name of the field in the child record where the value will be copied.",
+        examples=["parent_id", "subscription_updated"],
+        title="Target Field",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
@@ -2039,23 +2040,33 @@ class DefaultPaginator(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
-class DpathExtractor(BaseModel):
-    type: Literal["DpathExtractor"]
-    field_path: List[str] = Field(
+class RecordExpander(BaseModel):
+    type: Literal["RecordExpander"]
+    expand_records_from_field: List[str] = Field(
         ...,
-        description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).',
+        description="Path to a nested array field within each record. Items from this array will be extracted and emitted as separate records. Supports wildcards (*) for matching multiple arrays.",
         examples=[
-            ["data"],
-            ["data", "records"],
-            ["data", "{{ parameters.name }}"],
-            ["data", "*", "record"],
+            ["lines", "data"],
+            ["items"],
+            ["nested", "array"],
+            ["sections", "*", "items"],
         ],
-        title="Field Path",
+        title="Expand Records From Field",
     )
-    record_expander: Optional[RecordExpander] = Field(
+    remain_original_record: Optional[bool] = Field(
+        False,
+        description='If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.',
+        title="Remain Original Record",
+    )
+    on_no_records: Optional[OnNoRecords] = Field(
+        OnNoRecords.skip,
+        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.',
+        title="On No Records",
+    )
+    parent_fields_to_copy: Optional[List[ParentFieldMapping]] = Field(
         None,
-        description="Optional component to expand records by extracting items from nested array fields.",
-        title="Record Expander",
+        description="List of parent field mappings to copy onto each expanded child record. Each mapping specifies a source path in the parent record and a target field name in the child record.",
+        title="Parent Fields To Copy",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
@@ -2118,27 +2129,6 @@ class ListPartitionRouter(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
-class RecordSelector(BaseModel):
-    type: Literal["RecordSelector"]
-    extractor: Union[DpathExtractor, CustomRecordExtractor]
-    record_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field(
-        None,
-        description="Responsible for filtering records to be emitted by the Source.",
-        title="Record Filter",
-    )
-    schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field(
-        None,
-        description="Responsible for normalization according to the schema.",
-        title="Schema Normalization",
-    )
-    transform_before_filtering: Optional[bool] = Field(
-        None,
-        description="If true, transformation will be applied before record filtering.",
-        title="Transform Before Filtering",
-    )
-    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
-
-
 class PaginationReset(BaseModel):
     type: Literal["PaginationReset"]
     action: Action1
@@ -2259,6 +2249,27 @@ class Config:
     )
 
 
+class DpathExtractor(BaseModel):
+    type: Literal["DpathExtractor"]
+    field_path: List[str] = Field(
+        ...,
+        description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).',
+        examples=[
+            ["data"],
+            ["data", "records"],
+            ["data", "{{ parameters.name }}"],
+            ["data", "*", "record"],
+        ],
+        title="Field Path",
+    )
+    record_expander: Optional[RecordExpander] = Field(
+        None,
+        description="Optional component to expand records by extracting items from nested array fields.",
+        title="Record Expander",
+    )
+    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
+
+
 class ZipfileDecoder(BaseModel):
     class Config:
         extra = Extra.allow
@@ -2271,6 +2282,27 @@ class Config:
     )
 
 
+class RecordSelector(BaseModel):
+    type: Literal["RecordSelector"]
+    extractor: Union[DpathExtractor, CustomRecordExtractor]
+    record_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field(
+        None,
+        description="Responsible for filtering records to be emitted by the Source.",
+        title="Record Filter",
+    )
+    schema_normalization: Optional[Union[SchemaNormalization, CustomSchemaNormalization]] = Field(
+        None,
+        description="Responsible for normalization according to the schema.",
+        title="Schema Normalization",
+    )
+    transform_before_filtering: Optional[bool] = Field(
+        None,
+        description="If true, transformation will be applied before record filtering.",
+        title="Transform Before Filtering",
+    )
+    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
+
+
 class ConfigMigration(BaseModel):
     type: Literal["ConfigMigration"]
     description: Optional[str] = Field(
diff --git a/unit_tests/sources/declarative/extractors/test_dpath_extractor.py b/unit_tests/sources/declarative/extractors/test_dpath_extractor.py
