Move to a two-retriever instances approach

Author: lazebnyi

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -441,6 +441,7 @@
     AsyncRetriever,
     SimpleRetriever,
     SimpleRetrieverTestReadDecorator,
+    StateDelegatingRetriever,
 )
 from airbyte_cdk.sources.declarative.schema import (
     ComplexFieldType,
@@ -2759,15 +2760,24 @@ def create_state_delegating_retriever(
         stop_condition_on_cursor: bool = False,
         client_side_incremental_sync: Optional[Dict[str, Any]] = None,
         transformations: List[RecordTransformation],
-    ) -> Optional[SimpleRetriever]:
-        retriever_model = (
-            model.incremental_retriever
-            if self._connector_state_manager.get_stream_state(name, None)
-            else model.full_refresh_retriever
+    ) -> Optional[StateDelegatingRetriever]:
+        if not isinstance(stream_slicer, DatetimeBasedCursor) and not isinstance(stream_slicer, PerPartitionCursor):
+            raise ValueError("StateDelegatingRetriever requires a DatetimeBasedCursor")
+
+        full_refresh_retriever = self._create_component_from_model(
+            model=model.full_refresh_retriever,
+            config=config,
+            name=name,
+            primary_key=primary_key,
+            stream_slicer=stream_slicer,
+            request_options_provider=request_options_provider,
+            stop_condition_on_cursor=stop_condition_on_cursor,
+            client_side_incremental_sync=client_side_incremental_sync,
+            transformations=transformations,
         )
 
-        return self._create_component_from_model(  # type: ignore[no-any-return] # Will be created SimpleRetriever
-            model=retriever_model,
+        incremental_retriever = self._create_component_from_model(
+            model=model.incremental_retriever,
             config=config,
             name=name,
             primary_key=primary_key,
@@ -2778,6 +2788,13 @@ def create_state_delegating_retriever(
             transformations=transformations,
         )
 
+        return StateDelegatingRetriever(
+            full_data_retriever=full_refresh_retriever,
+            incremental_data_retriever=incremental_retriever,
+            cursor=stream_slicer,
+            started_with_state=bool(self._connector_state_manager.get_stream_state(name, None))
+        )
+
     def _create_async_job_status_mapping(
         self, model: AsyncJobStatusMapModel, config: Config, **kwargs: Any
     ) -> Mapping[str, AsyncJobStatus]:

airbyte_cdk/sources/declarative/retrievers/__init__.py

@@ -8,10 +8,14 @@
     SimpleRetriever,
     SimpleRetrieverTestReadDecorator,
 )
+from airbyte_cdk.sources.declarative.retrievers.state_delegating_retriever import (
+    StateDelegatingRetriever,
+)
 
 __all__ = [
     "Retriever",
     "SimpleRetriever",
     "SimpleRetrieverTestReadDecorator",
     "AsyncRetriever",
+    "StateDelegatingRetriever",
 ]

airbyte_cdk/sources/declarative/retrievers/state_delegating_retriever.py

@@ -0,0 +1,73 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass, field
+from typing import (
+    Any,
+    Mapping,
+)
+
+from typing_extensions import deprecated
+
+from airbyte_cdk.sources.declarative.incremental.declarative_cursor import DeclarativeCursor
+from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
+from airbyte_cdk.sources.source import ExperimentalClassWarning
+
+
+@deprecated(
+    "This class is experimental. Use at your own risk.",
+    category=ExperimentalClassWarning,
+)
+@dataclass
+class StateDelegatingRetriever:
+    full_data_retriever: Retriever
+    incremental_data_retriever: Retriever
+    cursor: DeclarativeCursor
+    started_with_state: bool = field(init=True, repr=False, default=False)
+
+    def __getattr__(self, name: str) -> Any:
+        # Avoid delegation for these internal names.
+        if name in {
+            "full_data_retriever",
+            "incremental_data_retriever",
+            "cursor",
+            "retriever",
+            "state",
+        }:
+            return object.__getattribute__(self, name)
+        # Delegate everything else to the active retriever.
+        return getattr(self.retriever, name)
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        # For the internal attributes, set them directly on self.
+        if name in {
+            "full_data_retriever",
+            "incremental_data_retriever",
+            "cursor",
+            "state",
+            "started_with_state",
+        }:
+            super().__setattr__(name, value)
+        else:
+            # Delegate setting attributes to the underlying retriever.
+            setattr(self.retriever, name, value)
+
+    @property
+    def retriever(self) -> Retriever:
+        return (
+            self.incremental_data_retriever
+            if self.started_with_state
+            else self.full_data_retriever
+        )
+
+    @property
+    def state(self) -> Mapping[str, Any]:
+        return self.cursor.get_stream_state() if self.cursor else {}
+
+    @state.setter
+    def state(self, value: Mapping[str, Any]) -> None:
+        """State setter, accept state serialized by state getter."""
+        if self.cursor:
+            self.cursor.set_initial_state(value)
+            self.started_with_state = bool(value)