refactor: Use stream cursor for retention period check, remove legacy get_cursor_datetime_from_state

- Rewrite create_state_delegating_stream to create actual stream object and extract cursor
- Add model-level check for IncrementingCountCursor with api_retention_period
- Delegate cursor datetime extraction to cursor's get_cursor_datetime_from_state method
- Remove get_cursor_datetime_from_state from legacy cursors (DeclarativeCursor, DatetimeBasedCursor, PerPartitionCursor)
- Remove factory helper methods (_create_cursor_for_age_check, _get_state_delegating_stream_model)
- Update tests to match new behavior

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

Author: devin-ai-integration[bot]

airbyte_cdk/legacy/sources/declarative/incremental/datetime_based_cursor.py

@@ -212,26 +212,6 @@ def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[S
         # through each slice and does not belong to a specific slice. We just return stream state as it is.
         return self.get_stream_state()
 
-    def get_cursor_datetime_from_state(
-        self, stream_state: Mapping[str, Any]
-    ) -> Optional[datetime.datetime]:
-        """Extract and parse the cursor datetime from the given stream state.
-
-        Returns the cursor datetime if present and parseable, otherwise returns None.
-        """
-        cursor_field_key = self.cursor_field.eval(self.config)  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
-        if cursor_field_key not in stream_state:
-            return None
-
-        cursor_value = stream_state.get(cursor_field_key)
-        if not cursor_value:
-            return None
-
-        try:
-            return self.parse_date(str(cursor_value))
-        except ValueError:
-            return None
-
     def _calculate_earliest_possible_value(
         self, end_datetime: datetime.datetime
     ) -> datetime.datetime:

airbyte_cdk/legacy/sources/declarative/incremental/declarative_cursor.py

@@ -1,8 +1,6 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 
-import datetime
 from abc import ABC
-from typing import Any, Mapping, Optional
 
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
 from airbyte_cdk.sources.streams.checkpoint.cursor import Cursor
@@ -13,24 +11,3 @@ class DeclarativeCursor(Cursor, StreamSlicer, ABC):
     DeclarativeCursors are components that allow for checkpointing syncs. In addition to managing the fetching and updating of
     state, declarative cursors also manage stream slicing and injecting slice values into outbound requests.
     """
-
-    def get_cursor_datetime_from_state(
-        self, stream_state: Mapping[str, Any]
-    ) -> Optional[datetime.datetime]:
-        """Extract and parse the cursor datetime from the given stream state.
-
-        This method is used by StateDelegatingStream to validate cursor age against
-        an API's data retention period. Subclasses should implement this method to
-        extract the cursor value from their specific state structure and parse it
-        into a datetime object.
-
-        Returns None if the cursor cannot be extracted or parsed, which will cause
-        StateDelegatingStream to fall back to full refresh (safe default).
-
-        Raises NotImplementedError by default - subclasses must implement this method
-        if they want to support cursor age validation with api_retention_period.
-        """
-        raise NotImplementedError(
-            f"{self.__class__.__name__} does not implement get_cursor_datetime_from_state. "
-            f"Cursor age validation with api_retention_period is not supported for this cursor type."
-        )

airbyte_cdk/legacy/sources/declarative/incremental/per_partition_cursor.py

@@ -2,7 +2,6 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-import datetime
 import logging
 from collections import OrderedDict
 from typing import Any, Callable, Iterable, Mapping, Optional, Union
@@ -212,23 +211,6 @@ def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[S
 
         return self._get_state_for_partition(stream_slice.partition)
 
-    def get_cursor_datetime_from_state(
-        self, stream_state: Mapping[str, Any]
-    ) -> Optional[datetime.datetime]:
-        """Extract and parse the cursor datetime from the global cursor in per-partition state.
-
-        For per-partition cursors, the global cursor is stored under the "state" key.
-        This method delegates to the underlying cursor factory to parse the datetime.
-
-        Returns None if the global cursor is not present or cannot be parsed.
-        """
-        global_state = stream_state.get("state")
-        if not global_state or not isinstance(global_state, dict):
-            return None
-
-        cursor = self._cursor_factory.create()
-        return cursor.get_cursor_datetime_from_state(global_state)
-
     def _create_cursor(self, cursor_state: Any) -> DeclarativeCursor:
         cursor = self._cursor_factory.create()
         cursor.set_initial_state(cursor_state)

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3565,86 +3565,70 @@ def create_state_delegating_stream(
                 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_model = self._get_state_delegating_stream_model(
-            False if has_parent_state is None else has_parent_state, model, config
-        )
-
-        return self._create_component_from_model(stream_model, config=config, **kwargs)  # type: ignore[no-any-return]  # DeclarativeStream will be created as stream_model is alwyas DeclarativeStreamModel
+        if model.api_retention_period:
+            for stream_model in (model.full_refresh_stream, model.incremental_stream):
+                if isinstance(stream_model.incremental_sync, IncrementingCountCursorModel):
+                    raise ValueError(
+                        f"Stream '{model.name}' uses IncrementingCountCursor which is not supported "
+                        f"with api_retention_period. IncrementingCountCursor does not use datetime-based "
+                        f"cursors, so cursor age validation cannot be performed."
+                    )
 
-    def _get_state_delegating_stream_model(
-        self, has_parent_state: bool, model: StateDelegatingStreamModel, config: Config
-    ) -> DeclarativeStreamModel:
         stream_state = self._connector_state_manager.get_stream_state(model.name, None)
+        has_parent = False if has_parent_state is None else has_parent_state
 
-        if not stream_state and not has_parent_state:
-            return model.full_refresh_stream
+        if not stream_state and not has_parent:
+            return self._create_component_from_model(model.full_refresh_stream, config=config, **kwargs)  # type: ignore[no-any-return]
+
+        incremental_stream: DefaultStream = self._create_component_from_model(model.incremental_stream, config=config, **kwargs)  # type: ignore[assignment]
 
         if model.api_retention_period and stream_state:
-            incremental_sync_sources = [
-                model.full_refresh_stream.incremental_sync,
-                model.incremental_stream.incremental_sync,
-            ]
-            incremental_sync_sources = [s for s in incremental_sync_sources if s is not None]
-            if incremental_sync_sources and self._is_cursor_older_than_retention_period(
-                stream_state,
-                incremental_sync_sources,
-                model.api_retention_period,
-                model.name,
-                config,
+            cursor = incremental_stream.cursor
+            if self._is_cursor_older_than_retention_period(
+                stream_state, cursor, model.api_retention_period, model.name
             ):
-                return model.full_refresh_stream
+                return self._create_component_from_model(model.full_refresh_stream, config=config, **kwargs)  # type: ignore[no-any-return]
 
-        return model.incremental_stream
+        return incremental_stream
 
+    @staticmethod
     def _is_cursor_older_than_retention_period(
-        self,
         stream_state: Mapping[str, Any],
-        incremental_sync_sources: list[Any],
+        cursor: Any,
         api_retention_period: str,
         stream_name: str,
-        config: Config,
     ) -> bool:
         """Check if the cursor value in the state is older than the API's retention period.
 
-        Delegates cursor datetime extraction to cursor class instances via
-        get_cursor_datetime_from_state, which handles format-specific parsing.
+        Delegates cursor datetime extraction to the cursor instance via
+        get_cursor_datetime_from_state.
 
         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).
-        Raises ValueError if the cursor datetime could not be parsed from state.
         """
-        # Skip retention check for concurrent state format (e.g. {"state_type": "date-range", "slices": [...]}).
-        # The DatetimeBasedCursor used for the age check only handles sequential state format.
-        # Today, is_sequential_state=True is hardcoded for all declarative cursors, so concurrent
-        # format state should never appear in practice. If that changes in the future, this guard
-        # prevents spurious full-refresh fallbacks until proper concurrent cursor delegation is added.
-        if "state_type" in stream_state or "slices" in stream_state:
-            return False
-
-        datetime_cursor_sources = [
-            s for s in incremental_sync_sources if isinstance(s, DatetimeBasedCursorModel)
-        ]
-        if not datetime_cursor_sources:
-            return False
+        if not hasattr(cursor, "get_cursor_datetime_from_state"):
+            raise SystemError(
+                f"Stream '{stream_name}' cursor type '{type(cursor).__name__}' does not have "
+                f"get_cursor_datetime_from_state method. Cursor age validation with "
+                f"api_retention_period is not supported for this cursor type."
+            )
 
-        cursor_datetime: datetime.datetime | None = None
-        for incremental_sync in datetime_cursor_sources:
-            cursor = self._create_cursor_for_age_check(incremental_sync, config)
+        try:
             cursor_datetime = cursor.get_cursor_datetime_from_state(stream_state)
-            if cursor_datetime is not None:
-                break
+        except NotImplementedError:
+            raise SystemError(
+                f"Stream '{stream_name}' cursor type '{type(cursor).__name__}' does not implement "
+                f"get_cursor_datetime_from_state. Cursor age validation with "
+                f"api_retention_period is not supported for this cursor type."
+            )
+
+        if cursor_datetime is None:
             global_state = stream_state.get("state")
             if isinstance(global_state, dict):
                 cursor_datetime = cursor.get_cursor_datetime_from_state(global_state)
-                if cursor_datetime is not None:
-                    break
 
         if cursor_datetime is None:
-            raise ValueError(
-                f"Stream '{stream_name}' has api_retention_period set to '{api_retention_period}' "
-                f"but the cursor datetime could not be parsed from state. Check that cursor_field "
-                f"and datetime_format match the state format."
-            )
+            return True
 
         retention_duration = parse_duration(api_retention_period)
         retention_cutoff = datetime.datetime.now(datetime.timezone.utc) - retention_duration
@@ -3660,24 +3644,6 @@ def _is_cursor_older_than_retention_period(
 
         return False
 
-    @staticmethod
-    def _create_cursor_for_age_check(
-        model: DatetimeBasedCursorModel, config: Config
-    ) -> "DatetimeBasedCursor":
-        """Create a lightweight DatetimeBasedCursor for cursor age validation."""
-        from airbyte_cdk.legacy.sources.declarative.incremental.datetime_based_cursor import (
-            DatetimeBasedCursor as _DatetimeBasedCursor,
-        )
-
-        return _DatetimeBasedCursor(
-            start_datetime="2000-01-01T00:00:00Z",
-            cursor_field=model.cursor_field,
-            datetime_format=model.datetime_format,
-            config=config,
-            parameters=model.parameters or {},
-            cursor_datetime_formats=model.cursor_datetime_formats or [],
-        )
-
     def _create_async_job_status_mapping(
         self, model: AsyncJobStatusMapModel, config: Config, **kwargs: Any
     ) -> Mapping[str, AsyncJobStatus]:

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -527,33 +527,26 @@ def _create_manifest_with_incrementing_count_cursor(api_retention_period: str) -
     return manifest
 
 
-def test_cursor_age_validation_skips_incrementing_count_cursor():
-    """Test that IncrementingCountCursor with api_retention_period is silently skipped (no error, uses incremental)."""
+def test_cursor_age_validation_raises_error_for_incrementing_count_cursor():
+    """Test that IncrementingCountCursor with api_retention_period raises ValueError."""
     manifest = _create_manifest_with_incrementing_count_cursor("P7D")
 
-    with HttpMocker() as http_mocker:
-        http_mocker.get(
-            HttpRequest(url="https://api.test.com/items_with_filtration"),
-            HttpResponse(body=json.dumps([{"id": 101, "updated_at": "2024-07-14"}])),
+    state = [
+        AirbyteStateMessage(
+            type=AirbyteStateType.STREAM,
+            stream=AirbyteStreamState(
+                stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                stream_state=AirbyteStateBlob(id=100),
+            ),
         )
+    ]
 
-        state = [
-            AirbyteStateMessage(
-                type=AirbyteStateType.STREAM,
-                stream=AirbyteStreamState(
-                    stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
-                    stream_state=AirbyteStateBlob(id=100),
-                ),
-            )
-        ]
-
-        source = ConcurrentDeclarativeSource(
-            source_config=manifest, config=_CONFIG, catalog=None, state=state
-        )
-        configured_catalog = create_configured_catalog(source, _CONFIG)
+    source = ConcurrentDeclarativeSource(
+        source_config=manifest, config=_CONFIG, catalog=None, state=state
+    )
 
-        records = get_records(source, _CONFIG, configured_catalog, state)
-        assert len(records) >= 0
+    with pytest.raises(ValueError, match="IncrementingCountCursor"):
+        source.discover(logger=MagicMock(), config=_CONFIG)
 
 
 def test_cursor_age_validation_raises_error_for_unparseable_cursor():
@@ -574,5 +567,5 @@ def test_cursor_age_validation_raises_error_for_unparseable_cursor():
         source_config=manifest, config=_CONFIG, catalog=None, state=state
     )
 
-    with pytest.raises(ValueError, match="could not be parsed from state"):
+    with pytest.raises(ValueError, match="not-a-date"):
         source.discover(logger=MagicMock(), config=_CONFIG)