Merge branch 'main' into daryna/low-code/refactor-model-to-component-factory

Author: darynaishchenko

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -108,7 +108,12 @@ def read_stream(
         stream_name = configured_catalog.streams[0].stream.name
 
         stream_read = test_read_handler.run_test_read(
-            source, config, configured_catalog, state, limits.max_records
+            source,
+            config,
+            configured_catalog,
+            stream_name,
+            state,
+            limits.max_records,
         )
 
         return AirbyteMessage(

airbyte_cdk/connector_builder/test_reader/helpers.py

@@ -269,6 +269,37 @@ 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.
+    """
+    if not json_message or not is_page_http_request(json_message):
+        return False
+
+    message_stream_name: str | None = (
+        json_message.get("airbyte_cdk", {}).get("stream", {}).get("name", None)
+    )
+    if message_stream_name is None:
+        return False
+
+    return message_stream_name != stream_name
+
+
 def is_page_http_request(json_message: Optional[Dict[str, Any]]) -> bool:
     """
     Determines whether a given JSON message represents a page HTTP request.

airbyte_cdk/connector_builder/test_reader/message_grouper.py

@@ -28,6 +28,7 @@
     is_async_auxiliary_request,
     is_config_update_message,
     is_log_message,
+    is_page_http_request_for_different_stream,
     is_record_message,
     is_state_message,
     is_trace_with_error,
@@ -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

@@ -86,6 +86,7 @@ def run_test_read(
         source: DeclarativeSource,
         config: Mapping[str, Any],
         configured_catalog: ConfiguredAirbyteCatalog,
+        stream_name: str,
         state: List[AirbyteStateMessage],
         record_limit: Optional[int] = None,
     ) -> StreamRead:
@@ -112,14 +113,17 @@ def run_test_read(
 
         record_limit = self._check_record_limit(record_limit)
         # The connector builder currently only supports reading from a single stream at a time
-        stream = source.streams(config)[0]
+        streams = source.streams(config)
+        stream = next((stream for stream in streams if stream.name == stream_name), None)
 
         # get any deprecation warnings during the component creation
         deprecation_warnings: List[LogMessage] = source.deprecation_warnings()
 
         schema_inferrer = SchemaInferrer(
-            self._pk_to_nested_and_composite_field(stream.primary_key),
-            self._cursor_field_to_nested_and_composite_field(stream.cursor_field),
+            self._pk_to_nested_and_composite_field(stream.primary_key) if stream else None,
+            self._cursor_field_to_nested_and_composite_field(stream.cursor_field)
+            if stream
+            else None,
         )
         datetime_format_inferrer = DatetimeFormatInferrer()
 
@@ -128,6 +132,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/concurrent_declarative_source.py

@@ -468,34 +468,11 @@ def _is_concurrent_cursor_incremental_without_partition_routing(
     def _get_retriever(
         declarative_stream: DeclarativeStream, stream_state: Mapping[str, Any]
     ) -> Retriever:
-        retriever = declarative_stream.retriever
-
-        # This is an optimization so that we don't invoke any cursor or state management flows within the
-        # low-code framework because state management is handled through the ConcurrentCursor.
-        if declarative_stream and isinstance(retriever, SimpleRetriever):
-            # Also a temporary hack. In the legacy Stream implementation, as part of the read,
-            # set_initial_state() is called to instantiate incoming state on the cursor. Although we no
-            # longer rely on the legacy low-code cursor for concurrent checkpointing, low-code components
-            # like StopConditionPaginationStrategyDecorator still rely on a DatetimeBasedCursor that is
-            # properly initialized with state.
-            if retriever.cursor:
-                retriever.cursor.set_initial_state(stream_state=stream_state)
-
-            # Similar to above, the ClientSideIncrementalRecordFilterDecorator cursor is a separate instance
-            # from the one initialized on the SimpleRetriever, so it also must also have state initialized
-            # for semi-incremental streams using is_client_side_incremental to filter properly
-            if isinstance(retriever.record_selector, RecordSelector) and isinstance(
-                retriever.record_selector.record_filter, ClientSideIncrementalRecordFilterDecorator
-            ):
-                retriever.record_selector.record_filter._cursor.set_initial_state(
-                    stream_state=stream_state
-                )  # type: ignore  # After non-concurrent cursors are deprecated we can remove these cursor workarounds
-
+        if declarative_stream and isinstance(declarative_stream.retriever, SimpleRetriever):
             # We zero it out here, but since this is a cursor reference, the state is still properly
             # instantiated for the other components that reference it
-            retriever.cursor = None
-
-        return retriever
+            declarative_stream.retriever.cursor = None
+        return declarative_stream.retriever
 
     @staticmethod
     def _select_streams(

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1511,7 +1511,6 @@ definitions:
       incremental_sync:
         title: Incremental Sync
         description: Component used to fetch data incrementally based on a time field in the data.
-        linkable: true
         anyOf:
           - "$ref": "#/definitions/DatetimeBasedCursor"
           - "$ref": "#/definitions/IncrementingCountCursor"
@@ -3682,7 +3681,6 @@ definitions:
       partition_router:
         title: Partition Router
         description: Used to iteratively execute requests over a set of values, such as a parent stream's records or a list of constant values.
-        linkable: true
         anyOf:
           - "$ref": "#/definitions/SubstreamPartitionRouter"
           - "$ref": "#/definitions/ListPartitionRouter"

airbyte_cdk/sources/declarative/extractors/record_filter.py

@@ -4,11 +4,6 @@
 from dataclasses import InitVar, dataclass
 from typing import Any, Callable, Iterable, Mapping, Optional, Union
 
-from airbyte_cdk.sources.declarative.incremental import (
-    DatetimeBasedCursor,
-    GlobalSubstreamCursor,
-    PerPartitionWithGlobalCursor,
-)
 from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     RecordFilter as RecordFilterModel,
@@ -17,6 +12,7 @@
     AdditionalFlags,
     ComponentConstructor,
 )
+from airbyte_cdk.sources.streams.concurrent.cursor import Cursor
 from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState
 
 
@@ -75,13 +71,13 @@ class ClientSideIncrementalRecordFilterDecorator(RecordFilter):
     """
     Applies a filter to a list of records to exclude those that are older than the stream_state/start_date.
 
-    :param DatetimeBasedCursor date_time_based_cursor: Cursor used to extract datetime values
+    :param Cursor cursor: Cursor used to filter out values
     :param PerPartitionCursor per_partition_cursor: Optional Cursor used for mapping cursor value in nested stream_state
     """
 
     def __init__(
         self,
-        cursor: Union[DatetimeBasedCursor, PerPartitionWithGlobalCursor, GlobalSubstreamCursor],
+        cursor: Union[Cursor],
         **kwargs: Any,
     ):
         super().__init__(**kwargs)
@@ -99,7 +95,7 @@ def filter_records(
             for record in records
             if self._cursor.should_be_synced(
                 # Record is created on the fly to align with cursors interface; stream name is ignored as we don't need it here
-                # Record stream name is empty cause it is not used durig the filtering
+                # Record stream name is empty because it is not used during the filtering
                 Record(data=record, associated_slice=stream_slice, stream_name="")
             )
         )

airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py

@@ -81,6 +81,7 @@ def __init__(
         connector_state_converter: AbstractStreamStateConverter,
         cursor_field: CursorField,
         use_global_cursor: bool = False,
+        attempt_to_create_cursor_if_not_provided: bool = False,
     ) -> None:
         self._global_cursor: Optional[StreamState] = {}
         self._stream_name = stream_name
@@ -125,6 +126,9 @@ def __init__(
 
         self._set_initial_state(stream_state)
 
+        # FIXME this is a temporary field the time of the migration from declarative cursors to concurrent ones
+        self._attempt_to_create_cursor_if_not_provided = attempt_to_create_cursor_if_not_provided
+
     @property
     def cursor_field(self) -> CursorField:
         return self._cursor_field
@@ -512,13 +516,28 @@ def _get_cursor(self, record: Record) -> ConcurrentCursor:
             raise ValueError(
                 "Invalid state as stream slices that are emitted should refer to an existing cursor"
             )
+
+        if self._use_global_cursor:
+            return self._create_cursor(
+                self._global_cursor,
+                self._lookback_window if self._global_cursor else 0,
+            )
+
         partition_key = self._to_partition_key(record.associated_slice.partition)
-        if partition_key not in self._cursor_per_partition:
+        if (
+            partition_key not in self._cursor_per_partition
+            and not self._attempt_to_create_cursor_if_not_provided
+        ):
             raise ValueError(
                 "Invalid state as stream slices that are emitted should refer to an existing cursor"
             )
-        cursor = self._cursor_per_partition[partition_key]
-        return cursor
+        elif partition_key not in self._cursor_per_partition:
+            return self._create_cursor(
+                self._global_cursor,
+                self._lookback_window if self._global_cursor else 0,
+            )
+        else:
+            return self._cursor_per_partition[partition_key]
 
     def limit_reached(self) -> bool:
         return self._number_of_partitions > self.SWITCH_TO_GLOBAL_LIMIT

airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -429,17 +429,6 @@ def _send_log(self, level: Level, message: str) -> None:
                 )
             )
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        cursor_field = self.cursor_field.eval(self.config)  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
-        first_cursor_value = first.get(cursor_field)
-        second_cursor_value = second.get(cursor_field)
-        if first_cursor_value and second_cursor_value:
-            return self.parse_date(first_cursor_value) >= self.parse_date(second_cursor_value)
-        elif first_cursor_value:
-            return True
-        else:
-            return False
-
     def set_runtime_lookback_window(self, lookback_window_in_seconds: int) -> None:
         """
         Updates the lookback window based on a given number of seconds if the new duration

airbyte_cdk/sources/declarative/incremental/global_substream_cursor.py

@@ -338,12 +338,6 @@ def get_request_body_json(
     def should_be_synced(self, record: Record) -> bool:
         return self._stream_cursor.should_be_synced(self._convert_record_to_cursor_record(record))
 
-    def is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
-        return self._stream_cursor.is_greater_than_or_equal(
-            self._convert_record_to_cursor_record(first),
-            self._convert_record_to_cursor_record(second),
-        )
-
     @staticmethod
     def _convert_record_to_cursor_record(record: Record) -> Record:
         return Record(