refactor: Clean NO_CURSOR_STATE_KEY from ConcurrentCursor, add tests for FinalStateCursor and parent retention

Co-Authored-By: gl_anatolii.yatsuk@airbyte.io <gl_anatolii.yatsuk@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -618,6 +618,7 @@
     NoopMessageRepository,
 )
 from airbyte_cdk.sources.message.repository import StateFilteringMessageRepository
+from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY
 from airbyte_cdk.sources.streams.call_rate import (
     APIBudget,
     FixedWindowCallRatePolicy,
@@ -3611,6 +3612,9 @@ def _is_cursor_older_than_retention_period(
         Returns True if the cursor is older than the retention period (should use full refresh).
         Returns False if the cursor is within the retention period (safe to use incremental).
         """
+        if stream_state.get(NO_CURSOR_STATE_KEY):
+            return False
+
         cursor_datetime: datetime.datetime | None = None
 
         for cursor in cursors:

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -608,9 +608,6 @@ def get_cursor_datetime_from_state(
 
         Returns the cursor datetime if present and parseable, otherwise returns None.
         """
-        if stream_state.get(NO_CURSOR_STATE_KEY):
-            return datetime.datetime.now(datetime.timezone.utc)
-
         # Check if state is in concurrent format (need to convert to dict for type compatibility)
         mutable_state: MutableMapping[str, Any] = dict(stream_state)
         if self._connector_state_converter.is_state_message_compatible(mutable_state):

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -603,3 +603,215 @@ def test_final_state_cursor_skips_retention_check_and_uses_incremental():
 
         records = get_records(source, _CONFIG, configured_catalog, state)
         assert len(records) == 1
+
+
+_PARENT_CHILD_MANIFEST: dict = {
+    "version": "6.0.0",
+    "type": "DeclarativeSource",
+    "check": {"type": "CheckStream", "stream_names": ["ChildStream"]},
+    "definitions": {
+        "ParentStream": {
+            "type": "StateDelegatingStream",
+            "name": "ParentStream",
+            "full_refresh_stream": {
+                "type": "DeclarativeStream",
+                "name": "ParentStream",
+                "primary_key": [],
+                "schema_loader": {
+                    "type": "InlineSchemaLoader",
+                    "schema": {
+                        "$schema": "http://json-schema.org/schema#",
+                        "properties": {},
+                        "type": "object",
+                    },
+                },
+                "retriever": {
+                    "type": "SimpleRetriever",
+                    "requester": {
+                        "type": "HttpRequester",
+                        "url_base": "https://api.test.com",
+                        "path": "/parents",
+                        "http_method": "GET",
+                    },
+                    "record_selector": {
+                        "type": "RecordSelector",
+                        "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_field": "updated_at",
+                },
+            },
+            "incremental_stream": {
+                "type": "DeclarativeStream",
+                "name": "ParentStream",
+                "primary_key": [],
+                "schema_loader": {
+                    "type": "InlineSchemaLoader",
+                    "schema": {
+                        "$schema": "http://json-schema.org/schema#",
+                        "properties": {},
+                        "type": "object",
+                    },
+                },
+                "retriever": {
+                    "type": "SimpleRetriever",
+                    "requester": {
+                        "type": "HttpRequester",
+                        "url_base": "https://api.test.com",
+                        "path": "/parents_incremental",
+                        "http_method": "GET",
+                    },
+                    "record_selector": {
+                        "type": "RecordSelector",
+                        "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",
+                    },
+                },
+            },
+        },
+        "ChildStream": {
+            "type": "DeclarativeStream",
+            "name": "ChildStream",
+            "primary_key": [],
+            "schema_loader": {
+                "type": "InlineSchemaLoader",
+                "schema": {
+                    "$schema": "http://json-schema.org/schema#",
+                    "properties": {},
+                    "type": "object",
+                },
+            },
+            "retriever": {
+                "type": "SimpleRetriever",
+                "requester": {
+                    "type": "HttpRequester",
+                    "url_base": "https://api.test.com",
+                    "path": "/children/{{ stream_slice.parent_id }}",
+                    "http_method": "GET",
+                },
+                "record_selector": {
+                    "type": "RecordSelector",
+                    "extractor": {"type": "DpathExtractor", "field_path": []},
+                },
+                "partition_router": {
+                    "type": "SubstreamPartitionRouter",
+                    "parent_stream_configs": [
+                        {
+                            "stream": "#/definitions/ParentStream",
+                            "parent_key": "id",
+                            "partition_field": "parent_id",
+                            "incremental_dependency": True,
+                        }
+                    ],
+                },
+            },
+            "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"],
+                "cursor_field": "updated_at",
+            },
+        },
+    },
+    "streams": [{"$ref": "#/definitions/ChildStream"}],
+    "spec": {
+        "connection_specification": {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "type": "object",
+            "required": [],
+            "properties": {},
+            "additionalProperties": True,
+        },
+        "documentation_url": "https://example.org",
+        "type": "Spec",
+    },
+}
+
+
+def _create_parent_child_manifest_with_retention_period(
+    api_retention_period: str,
+) -> dict:
+    manifest = copy.deepcopy(_PARENT_CHILD_MANIFEST)
+    manifest["definitions"]["ParentStream"]["api_retention_period"] = api_retention_period
+    return manifest
+
+
+@freezegun.freeze_time("2024-07-15")
+def test_parent_state_delegating_stream_retention_falls_back_to_full_refresh():
+    """When parent StateDelegatingStream has old cursor in child state, retention triggers full refresh for parent."""
+    manifest = _create_parent_child_manifest_with_retention_period("P7D")
+
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(url="https://api.test.com/parents"),
+            HttpResponse(
+                body=json.dumps(
+                    [{"id": 1, "name": "parent_1", "updated_at": "2024-07-14"}]
+                )
+            ),
+        )
+        http_mocker.get(
+            HttpRequest(url="https://api.test.com/children/1"),
+            HttpResponse(
+                body=json.dumps(
+                    [{"id": 10, "name": "child_1", "updated_at": "2024-07-14"}]
+                )
+            ),
+        )
+
+        state = [
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream=AirbyteStreamState(
+                    stream_descriptor=StreamDescriptor(
+                        name="ChildStream", namespace=None
+                    ),
+                    stream_state=AirbyteStateBlob(
+                        use_global_cursor=False,
+                        state={"updated_at": "2024-07-14"},
+                        states=[],
+                        parent_state={"ParentStream": {"updated_at": "2024-06-01"}},
+                        lookback_window=0,
+                    ),
+                ),
+            )
+        ]
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest, config=_CONFIG, catalog=None, state=state
+        )
+        configured_catalog = create_configured_catalog(source, _CONFIG)
+        records = get_records(source, _CONFIG, configured_catalog, state)
+        assert len(records) == 1

unit_tests/sources/streams/concurrent/test_cursor.py

@@ -13,6 +13,7 @@
 
 from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
 from airbyte_cdk.sources.message import MessageRepository
+from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY
 from airbyte_cdk.sources.streams.concurrent.clamping import (
     ClampingEndProvider,
     ClampingStrategy,
@@ -24,6 +25,7 @@
     ConcurrentCursor,
     CursorField,
     CursorValueType,
+    FinalStateCursor,
 )
 from airbyte_cdk.sources.streams.concurrent.partitions.partition import Partition
 from airbyte_cdk.sources.streams.concurrent.state_converters.abstract_stream_state_converter import (
@@ -1387,3 +1389,10 @@ def test_given_partitioned_state_with_multiple_slices_when_should_be_synced_then
         )
         == True
     )
+
+
+@freezegun.freeze_time("2024-07-15")
+def test_final_state_cursor_get_cursor_datetime_from_state_returns_current_datetime():
+    cursor = FinalStateCursor("test_stream", None, Mock(spec=MessageRepository))
+    result = cursor.get_cursor_datetime_from_state({NO_CURSOR_STATE_KEY: True})
+    assert result == datetime(2024, 7, 15, tzinfo=timezone.utc)