Update stream test

Author: lazebnyi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3160,12 +3160,20 @@ definitions:
     type: object
     required:
       - type
+      - name
       - full_refresh_stream
       - incremental_stream
     properties:
       type:
         type: string
         enum: [ StateDelegatingStream ]
+      name:
+        title: Name
+        description: The stream name.
+        type: string
+        default: ""
+        example:
+          - "Users"
       full_refresh_stream:
         title: Retriever
         description: Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2248,6 +2248,9 @@ class ParentStreamConfig(BaseModel):
 
 class StateDelegatingStream(BaseModel):
     type: Literal["StateDelegatingStream"]
+    name: str = Field(
+        ..., description="The stream name.", example=["Users"], title="Name"
+    )
     full_refresh_stream: DeclarativeStream = Field(
         ...,
         description="Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -355,9 +355,6 @@
     SimpleRetriever as SimpleRetrieverModel,
 )
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import Spec as SpecModel
-from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
-    StateDelegatingRetriever as StateDelegatingRetrieverModel,
-)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     StreamConfig as StreamConfigModel,
 )
@@ -444,7 +441,6 @@
     AsyncRetriever,
     SimpleRetriever,
     SimpleRetrieverTestReadDecorator,
-    StateDelegatingRetriever,
 )
 from airbyte_cdk.sources.declarative.schema import (
     ComplexFieldType,
@@ -624,7 +620,6 @@ def _init_mappings(self) -> None:
             SelectiveAuthenticatorModel: self.create_selective_authenticator,
             SimpleRetrieverModel: self.create_simple_retriever,
             StateDelegatingStreamModel: self.create_state_delegating_stream,
-            StateDelegatingRetrieverModel: self.create_state_delegating_retriever,
             SpecModel: self.create_spec,
             SubstreamPartitionRouterModel: self.create_substream_partition_router,
             WaitTimeFromHeaderModel: self.create_wait_time_from_header,
@@ -1785,7 +1780,6 @@ def _build_stream_slicer_from_partition_router(
             AsyncRetrieverModel,
             CustomRetrieverModel,
             SimpleRetrieverModel,
-            StateDelegatingRetrieverModel,
         ],
         config: Config,
         stream_name: Optional[str] = None,
@@ -1814,29 +1808,6 @@ def _build_incremental_cursor(
         stream_slicer: Optional[PartitionRouter],
         config: Config,
     ) -> Optional[StreamSlicer]:
-        if model.retriever.type == "StateDelegatingRetriever":
-            if not model.incremental_sync:
-                raise ValueError(
-                    "StateDelegatingRetriever requires 'incremental_sync' to be enabled."
-                )
-            elif model.incremental_sync.type != "DatetimeBasedCursor":
-                raise ValueError("StateDelegatingRetriever support only DatetimeBasedCursor.")
-            elif model.retriever.full_refresh_no_slice_in_params:
-                model.incremental_sync.step = None
-                model.incremental_sync.cursor_granularity = None
-                model.incremental_sync.start_time_option = None
-                model.incremental_sync.end_time_option = None
-            elif model.retriever.full_refresh_ignore_min_max_datetime:
-                start_datetime = model.incremental_sync.start_datetime
-                end_datetime = model.incremental_sync.end_datetime
-
-                if isinstance(start_datetime, MinMaxDatetimeModel):
-                    start_datetime.max_datetime = ""
-                    start_datetime.min_datetime = ""
-                if isinstance(end_datetime, MinMaxDatetimeModel):
-                    end_datetime.max_datetime = ""
-                    end_datetime.min_datetime = ""
-
         if model.incremental_sync and stream_slicer:
             if model.retriever.type == "AsyncRetriever":
                 return self.create_concurrent_cursor_from_perpartition_cursor(  # type: ignore # This is a known issue that we are creating and returning a ConcurrentCursor which does not technically implement the (low-code) StreamSlicer. However, (low-code) StreamSlicer and ConcurrentCursor both implement StreamSlicer.stream_slices() which is the primary method needed for checkpointing
@@ -1891,7 +1862,6 @@ def _build_resumable_cursor(
             AsyncRetrieverModel,
             CustomRetrieverModel,
             SimpleRetrieverModel,
-            StateDelegatingRetrieverModel,
         ],
         stream_slicer: Optional[PartitionRouter],
     ) -> Optional[StreamSlicer]:
@@ -1937,14 +1907,6 @@ def _merge_stream_slicers(
                 # we could support here by calling create_concurrent_cursor_from_perpartition_cursor
                 raise ValueError("Per partition state is not supported yet for AsyncRetriever.")
 
-        if retriever_model.type == "StateDelegatingRetriever":
-            stream_name = model.name or ""
-            retriever_model = (
-                retriever_model.incremental_retriever
-                if self._connector_state_manager.get_stream_state(stream_name, None)
-                else retriever_model.full_refresh_retriever
-            )
-
         stream_slicer = self._build_stream_slicer_from_partition_router(retriever_model, config)
 
         if model.incremental_sync:
@@ -2759,11 +2721,10 @@ def create_simple_retriever(
 
     def create_state_delegating_stream(self, model: StateDelegatingStreamModel, config: Config, child_state: Optional[MutableMapping[str, Any]] = None, **kwargs: Any
     ) -> DeclarativeStream:
-        if model.full_refresh_stream.name != model.incremental_stream.name:
-            raise ValueError(f"full_refresh_stream name and incremental_stream must have equal names. Instead has {model.full_refresh_stream.name}, {model.incremental_stream.name}.")
+        if model.full_refresh_stream.name != model.name or model.name != model.incremental_stream.name:
+            raise ValueError(f"state_delegating_stream, full_refresh_stream name and incremental_stream must have equal names. Instead has {model.name}, {model.full_refresh_stream.name} and {model.incremental_stream.name}.")
 
-        stream_name = model.full_refresh_stream.name
-        stream_model = model.incremental_stream if self._connector_state_manager.get_stream_state(stream_name, None) or child_state else model.full_refresh_stream
+        stream_model = model.incremental_stream if self._connector_state_manager.get_stream_state(model.name, None) or child_state else model.full_refresh_stream
 
         return self._create_component_from_model(stream_model, config=config, **kwargs)
 

…evers/test_state_delegating_retriever.py …larative/test_state_delegating_stream.pyunit_tests/sources/declarative/retrievers/test_state_delegating_retriever.py renamed to unit_tests/sources/declarative/test_state_delegating_stream.py unit_tests/sources/declarative/retrievers/test_state_delegating_retriever.py renamed to unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -30,20 +30,21 @@
     "check": {"type": "CheckStream", "stream_names": ["TestStream"]},
     "definitions": {
         "TestStream": {
-            "type": "DeclarativeStream",
+            "type": "StateDelegatingStream",
             "name": "TestStream",
-            "primary_key": [],
-            "schema_loader": {
-                "type": "InlineSchemaLoader",
-                "schema": {
-                    "$schema": "http://json-schema.org/schema#",
-                    "properties": {},
-                    "type": "object",
+            "full_refresh_stream": {
+                "type": "DeclarativeStream",
+                "name": "TestStream",
+                "primary_key": [],
+                "schema_loader": {
+                    "type": "InlineSchemaLoader",
+                    "schema": {
+                        "$schema": "http://json-schema.org/schema#",
+                        "properties": {},
+                        "type": "object",
+                    },
                 },
-            },
-            "retriever": {
-                "type": "StateDelegatingRetriever",
-                "full_refresh_retriever": {
+                "retriever": {
                     "type": "SimpleRetriever",
                     "requester": {
                         "type": "HttpRequester",
@@ -56,7 +57,30 @@
                         "extractor": {"type": "DpathExtractor", "field_path": []},
                     },
                 },
-                "incremental_retriever": {
+                "incremental_sync": {
+                    "type": "DatetimeBasedCursor",
+                    "start_datetime": {
+                        "datetime": "{{ format_datetime(config['start_date'], '%Y-%m-%d') }}"
+                    },
+                    "end_datetime": {"datetime": "{{ now_utc().strftime('%Y-%m-%d') }}"},
+                    "datetime_format": "%Y-%m-%d",
+                    "cursor_datetime_formats": ["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"],
+                    "cursor_field": "updated_at",
+                },
+            },
+            "incremental_stream": {
+                "type": "DeclarativeStream",
+                "name": "TestStream",
+                "primary_key": [],
+                "schema_loader": {
+                    "type": "InlineSchemaLoader",
+                    "schema": {
+                        "$schema": "http://json-schema.org/schema#",
+                        "properties": {},
+                        "type": "object",
+                    },
+                },
+                "retriever": {
                     "type": "SimpleRetriever",
                     "requester": {
                         "type": "HttpRequester",
@@ -69,29 +93,29 @@
                         "extractor": {"type": "DpathExtractor", "field_path": []},
                     },
                 },
-            },
-            "incremental_sync": {
-                "type": "DatetimeBasedCursor",
-                "start_datetime": {
-                    "datetime": "{{ format_datetime(config['start_date'], '%Y-%m-%d') }}"
-                },
-                "end_datetime": {"datetime": "{{ now_utc().strftime('%Y-%m-%d') }}"},
-                "datetime_format": "%Y-%m-%d",
-                "cursor_datetime_formats": ["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"],
-                "cursor_granularity": "P1D",
-                "step": "P15D",
-                "cursor_field": "updated_at",
-                "start_time_option": {
-                    "type": "RequestOption",
-                    "field_name": "start",
-                    "inject_into": "request_parameter",
-                },
-                "end_time_option": {
-                    "type": "RequestOption",
-                    "field_name": "end",
-                    "inject_into": "request_parameter",
+                "incremental_sync": {
+                    "type": "DatetimeBasedCursor",
+                    "start_datetime": {
+                        "datetime": "{{ format_datetime(config['start_date'], '%Y-%m-%d') }}"
+                    },
+                    "end_datetime": {"datetime": "{{ now_utc().strftime('%Y-%m-%d') }}"},
+                    "datetime_format": "%Y-%m-%d",
+                    "cursor_datetime_formats": ["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"],
+                    "cursor_granularity": "P1D",
+                    "step": "P15D",
+                    "cursor_field": "updated_at",
+                    "start_time_option": {
+                        "type": "RequestOption",
+                        "field_name": "start",
+                        "inject_into": "request_parameter",
+                    },
+                    "end_time_option": {
+                        "type": "RequestOption",
+                        "field_name": "end",
+                        "inject_into": "request_parameter",
+                    },
                 },
-            },
+            }
         },
     },
     "streams": [{"$ref": "#/definitions/TestStream"}],
@@ -166,7 +190,7 @@ def get_records(
 def test_state_retriever():
     with HttpMocker() as http_mocker:
         http_mocker.get(
-            HttpRequest(url="https://api.test.com/items?start=2024-07-01&end=2024-07-15"),
+            HttpRequest(url="https://api.test.com/items"),
             HttpResponse(
                 body=json.dumps(
                     [