check if page http request matches stream name

Author: lmossman

airbyte_cdk/connector_builder/test_reader/helpers.py

@@ -268,6 +268,25 @@ def should_close_page_for_slice(at_least_one_page_in_group: bool, message: Airby
     """
     return at_least_one_page_in_group and should_process_slice_descriptor(message)
 
+def is_page_http_request_for_different_stream(json_message: Optional[Dict[str, Any]], stream_name: str) -> bool:
+    """
+    Determines whether a given JSON message represents a page HTTP request for a different stream.
+
+    This function checks if the provided JSON message is a page HTTP request, and if the stream name in the log is
+    different from the provided stream name.
+
+    This is needed because dynamic streams result in extra page HTTP requests for the dynamic streams that we want to ignore
+    when they do not match the stream that is being read.
+
+    Args:
+        json_message (Optional[Dict[str, Any]]): The JSON message to evaluate.
+        stream_name (str): The name of the stream to compare against.
+
+    Returns:
+        bool: True if the JSON message is a page HTTP request for a different stream, False otherwise.
+    """
+    return is_page_http_request(json_message) and json_message.get("airbyte_cdk", {}).get("stream", {}).get("name", {}) != stream_name
+    
 
 def is_page_http_request(json_message: Optional[Dict[str, Any]]) -> bool:
     """

airbyte_cdk/connector_builder/test_reader/message_grouper.py

@@ -27,6 +27,7 @@
     handle_record_message,
     is_async_auxiliary_request,
     is_config_update_message,
+    is_page_http_request_for_different_stream,
     is_log_message,
     is_record_message,
     is_state_message,
@@ -44,6 +45,7 @@ def get_message_groups(
     schema_inferrer: SchemaInferrer,
     datetime_format_inferrer: DatetimeFormatInferrer,
     limit: int,
+    stream_name: str,
 ) -> MESSAGE_GROUPS:
     """
     Processes an iterator of AirbyteMessage objects to group and yield messages based on their type and sequence.
@@ -96,6 +98,9 @@ def get_message_groups(
     while records_count < limit and (message := next(messages, None)):
         json_message = airbyte_message_to_json(message)
 
+        if is_page_http_request_for_different_stream(json_message, stream_name):
+            continue
+
         if should_close_page(at_least_one_page_in_group, message, json_message):
             current_page_request, current_page_response = handle_current_page(
                 current_page_request,

airbyte_cdk/connector_builder/test_reader/reader.py

@@ -130,6 +130,7 @@ def run_test_read(
             schema_inferrer,
             datetime_format_inferrer,
             record_limit,
+            stream_name,
         )
 
         slices, log_messages, auxiliary_requests, latest_config_update = self._categorise_groups(

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -546,6 +546,7 @@ def _dynamic_stream_configs(
                 COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
                 components_resolver_config,
                 config,
+                stream_name=dynamic_definition.get("name"),
             )
 
             stream_template_config = dynamic_definition["stream_template"]

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3805,15 +3805,15 @@ def create_components_mapping_definition(
         )
 
     def create_http_components_resolver(
-        self, model: HttpComponentsResolverModel, config: Config
+        self, model: HttpComponentsResolverModel, config: Config, stream_name: Optional[str] = None
     ) -> Any:
         stream_slicer = self._build_stream_slicer_from_partition_router(model.retriever, config)
         combined_slicers = self._build_resumable_cursor(model.retriever, stream_slicer)
 
         retriever = self._create_component_from_model(
             model=model.retriever,
             config=config,
-            name="",
+            name=f"{stream_name if stream_name else '__http_components_resolver'}",
             primary_key=None,
             stream_slicer=stream_slicer if stream_slicer else combined_slicers,
             transformations=[],