Update parents resolving

Author: lazebnyi

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -29,6 +29,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DeclarativeStream as DeclarativeStreamModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    StateDelegatingStream as StateDelegatingStreamModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import Spec as SpecModel
 from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
     get_registered_components_module,
@@ -143,7 +146,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
 
         source_streams = [
             self._constructor.create_component(
-                DeclarativeStreamModel,
+                StateDelegatingStreamModel if stream_config.get("type") == StateDelegatingStreamModel.__name__ else DeclarativeStreamModel,
                 stream_config,
                 config,
                 emit_connector_builder_messages=self._emit_connector_builder_messages,
@@ -162,11 +165,11 @@ def _initialize_cache_for_parent_streams(
         def update_with_cache_parent_configs(parent_configs: list[dict[str, Any]]) -> None:
             for parent_config in parent_configs:
                 parent_streams.add(parent_config["stream"]["name"])
-                if parent_config["stream"]["retriever"]["type"] == "StateDelegatingRetriever":
-                    parent_config["stream"]["retriever"]["full_refresh_retriever"]["requester"][
+                if parent_config["stream"]["type"] == "StateDelegatingStream":
+                    parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"][
                         "use_cache"
                     ] = True
-                    parent_config["stream"]["retriever"]["incremental_retriever"]["requester"][
+                    parent_config["stream"]["incremental_stream"]["retriever"]["requester"][
                         "use_cache"
                     ] = True
                 else:
@@ -193,11 +196,11 @@ def update_with_cache_parent_configs(parent_configs: list[dict[str, Any]]) -> No
 
         for stream_config in stream_configs:
             if stream_config["name"] in parent_streams:
-                if stream_config["retriever"]["type"] == "StateDelegatingRetriever":
-                    stream_config["retriever"]["full_refresh_retriever"]["requester"][
+                if stream_config["type"] == "StateDelegatingStream":
+                    stream_config["full_refresh_stream"]["retriever"]["requester"][
                         "use_cache"
                     ] = True
-                    stream_config["retriever"]["incremental_retriever"]["requester"][
+                    stream_config["incremental_stream"]["retriever"]["requester"][
                         "use_cache"
                     ] = True
                 else: