Add RecordMergeStrategy interface and apply it to existing types

Author: brianjlai

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1032,6 +1032,9 @@ definitions:
       type:
         type: string
         enum: [EmitPartialRecordMergeStrategy]
+      $parameters:
+        type: object
+        additionalProperties: true
   JwtAuthenticator:
     title: JWT Authenticator
     description: Authenticator for requests using JWT authentication flow.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -347,6 +347,7 @@ class Clamping(BaseModel):
 
 class EmitPartialRecordMergeStrategy(BaseModel):
     type: Literal["EmitPartialRecordMergeStrategy"]
+    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
 class Algorithm(Enum):

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -218,6 +218,10 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DynamicSchemaLoader as DynamicSchemaLoaderModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    EmitPartialRecordMergeStrategy,
+    ValueType,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ExponentialBackoffStrategy as ExponentialBackoffStrategyModel,
 )
@@ -384,7 +388,6 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     UnlimitedCallRatePolicy as UnlimitedCallRatePolicyModel,
 )
-from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     WaitTimeFromHeader as WaitTimeFromHeaderModel,
 )
@@ -441,14 +444,17 @@
     StopConditionPaginationStrategyDecorator,
 )
 from airbyte_cdk.sources.declarative.requesters.query_properties import (
-    GroupByKey,
     PropertiesFromEndpoint,
     PropertyChunking,
     QueryProperties,
 )
 from airbyte_cdk.sources.declarative.requesters.query_properties.property_chunking import (
     PropertyLimitType,
 )
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies import (
+    EmitPartialRecord,
+    GroupByKey,
+)
 from airbyte_cdk.sources.declarative.requesters.request_option import RequestOptionType
 from airbyte_cdk.sources.declarative.requesters.request_options import (
     DatetimeBasedRequestOptionsProvider,
@@ -609,6 +615,7 @@ def _init_mappings(self) -> None:
             DefaultErrorHandlerModel: self.create_default_error_handler,
             DefaultPaginatorModel: self.create_default_paginator,
             DpathExtractorModel: self.create_dpath_extractor,
+            EmitPartialRecordMergeStrategy: self.create_emit_partial_record,
             ResponseToFileExtractorModel: self.create_response_to_file_extractor,
             ExponentialBackoffStrategyModel: self.create_exponential_backoff_strategy,
             SessionTokenAuthenticatorModel: self.create_session_token_authenticator,
@@ -790,6 +797,12 @@ def create_dpath_flatten_fields(
             parameters=model.parameters or {},
         )
 
+    @staticmethod
+    def create_emit_partial_record(
+        model: EmitPartialRecordMergeStrategy, config: Config, **kwargs: Any
+    ) -> EmitPartialRecord:
+        return EmitPartialRecord(config=config, parameters=model.parameters or {})
+
     @staticmethod
     def _json_schema_type_name_to_type(value_type: Optional[ValueType]) -> Optional[Type[Any]]:
         if not value_type:

airbyte_cdk/sources/declarative/requesters/query_properties/__init__.py

@@ -1,6 +1,5 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 
-from airbyte_cdk.sources.declarative.requesters.query_properties.group_by_key import GroupByKey
 from airbyte_cdk.sources.declarative.requesters.query_properties.properties_from_endpoint import (
     PropertiesFromEndpoint,
 )
@@ -11,4 +10,4 @@
     QueryProperties,
 )
 
-__all__ = ["GroupByKey", "PropertiesFromEndpoint", "PropertyChunking", "QueryProperties"]
+__all__ = ["PropertiesFromEndpoint", "PropertyChunking", "QueryProperties"]

airbyte_cdk/sources/declarative/requesters/query_properties/property_chunking.py

@@ -4,7 +4,10 @@
 from enum import Enum
 from typing import Any, Iterable, List, Mapping, Optional
 
-from airbyte_cdk.sources.declarative.requesters.query_properties import GroupByKey
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies import GroupByKey
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies.merge_strategy import (
+    RecordMergeStrategy,
+)
 from airbyte_cdk.sources.types import Config, Record
 
 
@@ -26,9 +29,7 @@ class PropertyChunking:
 
     property_limit_type: PropertyLimitType
     property_limit: Optional[int]
-    record_merge_strategy: Optional[
-        GroupByKey
-    ]  # This should eventually be some sort of interface or type
+    record_merge_strategy: Optional[RecordMergeStrategy]
     parameters: InitVar[Mapping[str, Any]]
     config: Config
 

airbyte_cdk/sources/declarative/requesters/query_properties/strategies/__init__.py

@@ -0,0 +1,13 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies.emit_partial_record import (
+    EmitPartialRecord,
+)
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies.group_by_key import (
+    GroupByKey,
+)
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies.merge_strategy import (
+    RecordMergeStrategy,
+)
+
+__all__ = ["EmitPartialRecord", "GroupByKey", "RecordMergeStrategy"]

airbyte_cdk/sources/declarative/requesters/query_properties/strategies/emit_partial_record.py

@@ -0,0 +1,23 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+from dataclasses import InitVar, dataclass
+from typing import Any, Mapping, Optional
+
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies.merge_strategy import (
+    RecordMergeStrategy,
+)
+from airbyte_cdk.sources.types import Config, Record
+
+
+@dataclass
+class EmitPartialRecord(RecordMergeStrategy):
+    """
+    Record merge strategy that emits partial records as they are without merging them together usually if
+    there is not a suitable primary key to merge on.
+    """
+
+    parameters: InitVar[Mapping[str, Any]]
+    config: Config
+
+    def get_group_key(self, record: Record) -> Optional[str]:
+        return None

…uesters/query_properties/group_by_key.py …ry_properties/strategies/group_by_key.pyairbyte_cdk/sources/declarative/requesters/query_properties/group_by_key.py renamed to airbyte_cdk/sources/declarative/requesters/query_properties/strategies/group_by_key.py airbyte_cdk/sources/declarative/requesters/query_properties/group_by_key.py renamed to airbyte_cdk/sources/declarative/requesters/query_properties/strategies/group_by_key.py

@@ -3,11 +3,14 @@
 from dataclasses import InitVar, dataclass
 from typing import Any, List, Mapping, Optional, Union
 
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies.merge_strategy import (
+    RecordMergeStrategy,
+)
 from airbyte_cdk.sources.types import Config, Record
 
 
 @dataclass
-class GroupByKey:
+class GroupByKey(RecordMergeStrategy):
     """
     Record merge strategy that combines records together according to values on the record for one or many keys.
     """

airbyte_cdk/sources/declarative/requesters/query_properties/strategies/merge_strategy.py

@@ -0,0 +1,19 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Optional
+
+from airbyte_cdk.sources.types import Record
+
+
+@dataclass
+class RecordMergeStrategy(ABC):
+    """
+    Describe the interface for how records that required multiple requests to get the complete set of fields
+    should be merged back into a single record.
+    """
+
+    @abstractmethod
+    def get_group_key(self, record: Record) -> Optional[str]:
+        pass

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -9,7 +9,6 @@
 import freezegun
 import pytest
 import requests
-from pydantic.v1 import ValidationError
 
 from airbyte_cdk import AirbyteTracedException
 from airbyte_cdk.models import (
@@ -127,14 +126,14 @@
     StopConditionPaginationStrategyDecorator,
 )
 from airbyte_cdk.sources.declarative.requesters.query_properties import (
-    GroupByKey,
     PropertiesFromEndpoint,
     PropertyChunking,
     QueryProperties,
 )
 from airbyte_cdk.sources.declarative.requesters.query_properties.property_chunking import (
     PropertyLimitType,
 )
+from airbyte_cdk.sources.declarative.requesters.query_properties.strategies import GroupByKey
 from airbyte_cdk.sources.declarative.requesters.request_option import (
     RequestOption,
     RequestOptionType,
@@ -4376,7 +4375,7 @@ def test_create_property_chunking_invalid_property_limit_type():
 
     connector_builder_factory = ModelToComponentFactory(emit_connector_builder_messages=True)
 
-    with pytest.raises(ValidationError):
+    with pytest.raises:
         connector_builder_factory.create_component(
             model_type=PropertyChunkingModel,
             component_definition=property_chunking_model,