feat(low-code cdk): add StateDelegatingStream

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1311,6 +1311,7 @@ definitions:
           - "$ref": "#/definitions/AsyncRetriever"
           - "$ref": "#/definitions/CustomRetriever"
           - "$ref": "#/definitions/SimpleRetriever"
+          - "$ref": "#/definitions/StateDelegatingRetriever"
       incremental_sync:
         title: Incremental Sync
         description: Component used to fetch data incrementally based on a time field in the data.
@@ -2955,6 +2956,34 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  StateDelegatingRetriever:
+    description: Test state condition retriever.
+    type: object
+    required:
+      - type
+      - incremental_data_retriever
+      - full_data_retriever
+    properties:
+      type:
+        type: string
+        enum: [ StateDelegatingRetriever ]
+      incremental_data_retriever:
+        title: Retriever
+        description: Component used to coordinate how records are extracted across stream slices and request pages.
+        anyOf:
+          - "$ref": "#/definitions/AsyncRetriever"
+          - "$ref": "#/definitions/CustomRetriever"
+          - "$ref": "#/definitions/SimpleRetriever"
+      full_data_retriever:
+        title: Retriever
+        description: Component used to coordinate how records are extracted across stream slices and request pages.
+        anyOf:
+          - "$ref": "#/definitions/AsyncRetriever"
+          - "$ref": "#/definitions/CustomRetriever"
+          - "$ref": "#/definitions/SimpleRetriever"
+      $parameters:
+        type: object
+        additionalProperties: true
   SimpleRetriever:
     description: Retrieves records by synchronously sending requests to fetch records. The retriever acts as an orchestrator between the requester, the record selector, the paginator, and the partition router.
     type: object

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1841,10 +1841,12 @@ class Config:
         extra = Extra.allow
 
     type: Literal["DeclarativeStream"]
-    retriever: Union[AsyncRetriever, CustomRetriever, SimpleRetriever] = Field(
-        ...,
-        description="Component used to coordinate how records are extracted across stream slices and request pages.",
-        title="Retriever",
+    retriever: Union[AsyncRetriever, CustomRetriever, SimpleRetriever, StateDelegatingRetriever] = (
+        Field(
+            ...,
+            description="Component used to coordinate how records are extracted across stream slices and request pages.",
+            title="Retriever",
+        )
     )
     incremental_sync: Optional[Union[CustomIncrementalSync, DatetimeBasedCursor]] = Field(
         None,
@@ -2100,6 +2102,21 @@ class ParentStreamConfig(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
+class StateDelegatingRetriever(BaseModel):
+    type: Literal["StateDelegatingRetriever"]
+    incremental_data_retriever: Union[AsyncRetriever, CustomRetriever, SimpleRetriever] = Field(
+        ...,
+        description="Component used to coordinate how records are extracted across stream slices and request pages.",
+        title="Retriever",
+    )
+    full_data_retriever: Union[AsyncRetriever, CustomRetriever, SimpleRetriever] = Field(
+        ...,
+        description="Component used to coordinate how records are extracted across stream slices and request pages.",
+        title="Retriever",
+    )
+    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
+
+
 class SimpleRetriever(BaseModel):
     type: Literal["SimpleRetriever"]
     record_selector: RecordSelector = Field(
@@ -2284,5 +2301,6 @@ class DynamicDeclarativeStream(BaseModel):
 DeclarativeStream.update_forward_refs()
 SessionTokenAuthenticator.update_forward_refs()
 DynamicSchemaLoader.update_forward_refs()
+StateDelegatingRetriever.update_forward_refs()
 SimpleRetriever.update_forward_refs()
 AsyncRetriever.update_forward_refs()

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -347,6 +347,9 @@
     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,
 )
@@ -430,6 +433,7 @@
     AsyncRetriever,
     SimpleRetriever,
     SimpleRetrieverTestReadDecorator,
+    StateDelegatingRetriever,
 )
 from airbyte_cdk.sources.declarative.schema import (
     ComplexFieldType,
@@ -597,6 +601,7 @@ def _init_mappings(self) -> None:
             LegacySessionTokenAuthenticatorModel: self.create_legacy_session_token_authenticator,
             SelectiveAuthenticatorModel: self.create_selective_authenticator,
             SimpleRetrieverModel: self.create_simple_retriever,
+            StateDelegatingRetrieverModel: self.create_state_delegating_retriever,
             SpecModel: self.create_spec,
             SubstreamPartitionRouterModel: self.create_substream_partition_router,
             WaitTimeFromHeaderModel: self.create_wait_time_from_header,
@@ -2543,6 +2548,52 @@ def create_simple_retriever(
             parameters=model.parameters or {},
         )
 
+    def create_state_delegating_retriever(
+        self,
+        model: StateDelegatingRetrieverModel,
+        config: Config,
+        *,
+        name: str,
+        primary_key: Optional[Union[str, List[str], List[List[str]]]],
+        stream_slicer: Optional[StreamSlicer],
+        request_options_provider: Optional[RequestOptionsProvider] = None,
+        stop_condition_on_cursor: bool = False,
+        client_side_incremental_sync: Optional[Dict[str, Any]] = None,
+        transformations: List[RecordTransformation],
+    ) -> StateDelegatingRetriever:
+        if not isinstance(stream_slicer, DeclarativeCursor):
+            raise ValueError("StateDelegatingRetriever requires a DeclarativeCursor")
+
+        full_data_retriever = self._create_component_from_model(
+            model=model.full_data_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,
+        )
+
+        incremental_data_retriever = self._create_component_from_model(
+            model=model.incremental_data_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 StateDelegatingRetriever(
+            full_data_retriever=full_data_retriever,
+            incremental_data_retriever=incremental_data_retriever,
+            cursor=stream_slicer,
+        )
+
     def _create_async_job_status_mapping(
         self, model: AsyncJobStatusMapModel, config: Config, **kwargs: Any
     ) -> Mapping[str, AsyncJobStatus]:

airbyte_cdk/sources/declarative/retrievers/__init__.py

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

airbyte_cdk/sources/declarative/retrievers/state_delegating_retriever.py

@@ -0,0 +1,65 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import (
+    Any,
+    MutableMapping,
+)
+
+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
+
+    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"}:
+            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.cursor.get_stream_state()
+            else self.full_data_retriever
+        )
+
+    @property
+    def state(self) -> MutableMapping[str, Any]:
+        return self.cursor.get_stream_state() if self.cursor else {}
+
+    @state.setter
+    def state(self, value: MutableMapping[str, Any]) -> None:
+        """State setter, accept state serialized by state getter."""
+        if self.cursor:
+            self.cursor.set_initial_state(value)