refactor(RecordExpander): Remove ParentFieldMapping to delegate to RecordTransformations

Author: agarctfi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1969,46 +1969,6 @@ definitions:
           - 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

airbyte_cdk/sources/declarative/expanders/__init__.py

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

airbyte_cdk/sources/declarative/expanders/record_expander.py

@@ -2,49 +2,23 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from dataclasses import InitVar, dataclass, field
-from typing import Any, Iterable, Mapping, MutableMapping, Sequence
+from dataclasses import InitVar, dataclass
+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: Sequence[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.
 
     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.
+    within each record and emits each item as a separate record. Set `remain_original_record: true`
+    to embed the full parent record under `original_record` in each expanded item when you need
+    downstream transformations to access parent context.
 
     The expand_records_from_field path supports wildcards (*) for matching multiple arrays.
     When wildcards are used, items from all matched arrays are extracted and emitted.
@@ -67,10 +41,6 @@ class RecordExpander:
           - "*"
           - "items"
         on_no_records: emit_parent
-        parent_fields_to_copy:
-          - type: ParentFieldMapping
-            source_field_path: ["id"]
-            target_field: "parent_id"
     ```
 
     Attributes:
@@ -81,8 +51,6 @@ class RecordExpander:
             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.
     """
 
@@ -91,7 +59,6 @@ class RecordExpander:
     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: list[InterpolatedString] | None = [
@@ -148,6 +115,3 @@ def _apply_parent_context(
         """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)

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -492,23 +492,6 @@ class OnNoRecords(Enum):
     emit_parent = "emit_parent"
 
 
-class ParentFieldMapping(BaseModel):
-    type: Literal["ParentFieldMapping"]
-    source_field_path: List[str] = Field(
-        ...,
-        description="Path to the field in the parent record to copy.",
-        examples=[["id"], ["created"], ["metadata", "timestamp"]],
-        title="Source Field Path",
-    )
-    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")
-
-
 class ExponentialBackoffStrategy(BaseModel):
     type: Literal["ExponentialBackoffStrategy"]
     factor: Optional[Union[float, str]] = Field(
@@ -2063,11 +2046,6 @@ class RecordExpander(BaseModel):
         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="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")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -104,6 +104,7 @@
     RecordSelector,
     ResponseToFileExtractor,
 )
+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.extractors.record_filter import (
     ClientSideIncrementalRecordFilterDecorator,
@@ -374,9 +375,6 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ParametrizedComponentsResolver as ParametrizedComponentsResolverModel,
 )
-from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
-    ParentFieldMapping as ParentFieldMappingModel,
-)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ParentStreamConfig as ParentStreamConfigModel,
 )
@@ -2347,25 +2345,12 @@ def create_record_expander(
         config: Config,
         **kwargs: Any,
     ) -> RecordExpander:
-        parent_field_mappings: list[ParentFieldMapping] = []
-        if model.parent_fields_to_copy:
-            for mapping_model in model.parent_fields_to_copy:
-                parent_field_mappings.append(
-                    ParentFieldMapping(
-                        source_field_path=mapping_model.source_field_path,
-                        target_field=mapping_model.target_field,
-                        config=config,
-                        parameters=mapping_model.parameters or {},
-                    )
-                )
-
         return RecordExpander(
             expand_records_from_field=model.expand_records_from_field,
             config=config,
             parameters=model.parameters or {},
             remain_original_record=model.remain_original_record or False,
             on_no_records=model.on_no_records.value if model.on_no_records else "skip",
-            parent_fields_to_copy=parent_field_mappings,
         )
 
     @staticmethod