pr feedback to get rid of emit strategy, add caching

Author: brianjlai

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -341,7 +341,7 @@ definitions:
     properties:
       type:
         type: string
-        enum: [ DynamicStreamCheckConfig ]
+        enum: [DynamicStreamCheckConfig]
       dynamic_stream_name:
         title: Dynamic Stream Name
         description: The dynamic stream name.
@@ -1044,18 +1044,6 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
-  EmitPartialRecordMergeStrategy:
-    title: Emit Partial Record
-    description: Record merge strategy where in the case where multiple requests are needed to retrieve all properties, properties are not consolidated back into a single record and are instead emitted as separate groups of properties. This strategy should only be used when records do not have a unique identifier like a primary key.
-    required:
-      - type
-    properties:
-      type:
-        type: string
-        enum: [EmitPartialRecordMergeStrategy]
-      $parameters:
-        type: object
-        additionalProperties: true
   JwtAuthenticator:
     title: JWT Authenticator
     description: Authenticator for requests using JWT authentication flow.
@@ -3081,9 +3069,7 @@ definitions:
       record_merge_strategy:
         title: Record Merge Strategy
         description: Dictates how to records that require multiple requests to get all properties should be emitted to the destination
-        anyOf:
-          - "$ref": "#/definitions/EmitPartialRecordMergeStrategy"
-          - "$ref": "#/definitions/GroupByKeyMergeStrategy"
+        "$ref": "#/definitions/GroupByKeyMergeStrategy"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -51,7 +51,7 @@ class DynamicStreamCheckConfig(BaseModel):
     )
     stream_count: Optional[int] = Field(
         0,
-        description="Numbers of the streams to try reading from when running a check operation.",
+        description="The number of streams to attempt reading from during a check operation. If `stream_count` exceeds the total number of available streams, the minimum of the two values will be used.",
         title="Stream Count",
     )
 
@@ -347,11 +347,6 @@ class Clamping(BaseModel):
     target_details: Optional[Dict[str, Any]] = None
 
 
-class EmitPartialRecordMergeStrategy(BaseModel):
-    type: Literal["EmitPartialRecordMergeStrategy"]
-    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
-
-
 class Algorithm(Enum):
     HS256 = "HS256"
     HS384 = "HS384"
@@ -1224,9 +1219,7 @@ class PropertyChunking(BaseModel):
         description="The maximum amount of properties that can be retrieved per request according to the limit type.",
         title="Property Limit",
     )
-    record_merge_strategy: Optional[
-        Union[EmitPartialRecordMergeStrategy, GroupByKeyMergeStrategy]
-    ] = Field(
+    record_merge_strategy: Optional[GroupByKeyMergeStrategy] = Field(
         None,
         description="Dictates how to records that require multiple requests to get all properties should be emitted to the destination",
         title="Record Merge Strategy",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -225,9 +225,6 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DynamicStreamCheckConfig as DynamicStreamCheckConfigModel,
 )
-from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
-    EmitPartialRecordMergeStrategy as EmitPartialRecordMergeStrategyModel,
-)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ExponentialBackoffStrategy as ExponentialBackoffStrategyModel,
 )
@@ -459,7 +456,6 @@
     PropertyLimitType,
 )
 from airbyte_cdk.sources.declarative.requesters.query_properties.strategies import (
-    EmitPartialRecord,
     GroupByKey,
 )
 from airbyte_cdk.sources.declarative.requesters.request_option import RequestOptionType
@@ -623,7 +619,6 @@ def _init_mappings(self) -> None:
             DefaultErrorHandlerModel: self.create_default_error_handler,
             DefaultPaginatorModel: self.create_default_paginator,
             DpathExtractorModel: self.create_dpath_extractor,
-            EmitPartialRecordMergeStrategyModel: self.create_emit_partial_record,
             ResponseToFileExtractorModel: self.create_response_to_file_extractor,
             ExponentialBackoffStrategyModel: self.create_exponential_backoff_strategy,
             SessionTokenAuthenticatorModel: self.create_session_token_authenticator,
@@ -805,12 +800,6 @@ def create_dpath_flatten_fields(
             parameters=model.parameters or {},
         )
 
-    @staticmethod
-    def create_emit_partial_record(
-        model: EmitPartialRecordMergeStrategyModel, 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:
@@ -2149,6 +2138,7 @@ def create_http_requester(
         config: Config,
         decoder: Decoder = JsonDecoder(parameters={}),
         query_properties_key: Optional[str] = None,
+        use_cache: Optional[bool] = None,
         *,
         name: str,
     ) -> HttpRequester:
@@ -2189,7 +2179,7 @@ def create_http_requester(
         assert model.use_cache is not None  # for mypy
         assert model.http_method is not None  # for mypy
 
-        use_cache = model.use_cache and not self._disable_cache
+        should_use_cache = (model.use_cache or bool(use_cache)) and not self._disable_cache
 
         return HttpRequester(
             name=name,
@@ -2204,7 +2194,7 @@ def create_http_requester(
             disable_retries=self._disable_retries,
             parameters=model.parameters or {},
             message_repository=self._message_repository,
-            use_cache=use_cache,
+            use_cache=should_use_cache,
             decoder=decoder,
             stream_response=decoder.is_stream_response() if decoder else False,
         )
@@ -2308,10 +2298,11 @@ def create_dynamic_schema_loader(
         retriever = self._create_component_from_model(
             model=model.retriever,
             config=config,
-            name="",
+            name="dynamic_properties",
             primary_key=None,
             stream_slicer=combined_slicers,
             transformations=[],
+            use_cache=True,
         )
         schema_type_identifier = self._create_component_from_model(
             model.schema_type_identifier, config=config, parameters=model.parameters or {}
@@ -2652,14 +2643,14 @@ def create_parent_stream_config(
     def create_properties_from_endpoint(
         self, model: PropertiesFromEndpointModel, config: Config, **kwargs: Any
     ) -> PropertiesFromEndpoint:
-        name = "property_retriever"
         retriever = self._create_component_from_model(
             model=model.retriever,
             config=config,
-            name=name,
+            name="dynamic_properties",
             primary_key=None,
             stream_slicer=None,
             transformations=[],
+            use_cache=True,  # Enable caching on the HttpRequester/HttpClient because the properties endpoint will be called for every slice being processed, and it is highly unlikely for the response to different
         )
         return PropertiesFromEndpoint(
             property_field_path=model.property_field_path,
@@ -2867,6 +2858,7 @@ def create_simple_retriever(
                 IncrementingCountCursorModel, DatetimeBasedCursorModel, CustomIncrementalSyncModel
             ]
         ] = None,
+        use_cache: Optional[bool] = None,
         **kwargs: Any,
     ) -> SimpleRetriever:
         decoder = (
@@ -2928,9 +2920,10 @@ def create_simple_retriever(
         requester = self._create_component_from_model(
             model=model.requester,
             decoder=decoder,
+            name=name,
             query_properties_key=query_properties_key,
+            use_cache=use_cache,
             config=config,
-            name=name,
         )
         url_base = (
             model.requester.url_base

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

@@ -28,6 +28,12 @@ class QueryProperties:
     def get_request_property_chunks(
         self, stream_slice: Optional[StreamSlice] = None
     ) -> Iterable[List[str]]:
+        """
+        Uses the defined property_list to fetch the total set of properties dynamically or from a static list
+        and based on the resulting properties, performs property chunking if applicable.
+        :param stream_slice: The StreamSlice of the current partition being processed during the sync. This is included
+        because subcomponents of QueryProperties can make use of interpolation of the top-level StreamSlice object
+        """
         fields: Union[Iterable[str], List[str]]
         if isinstance(self.property_list, PropertiesFromEndpoint):
             fields = self.property_list.get_properties_from_endpoint(stream_slice=stream_slice)

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

@@ -1,13 +1,10 @@
 # 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"]
+__all__ = ["GroupByKey", "RecordMergeStrategy"]