Merge master to branch

Author: lazebnyi

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -217,7 +217,11 @@ def _group_streams(
                     declarative_stream.name
                 ].get("incremental_sync")
 
-                if name_to_stream_mapping[declarative_stream.name].get("retriever", {}).get("full_refresh_no_slice_in_params", False):
+                if (
+                    name_to_stream_mapping[declarative_stream.name]
+                    .get("retriever", {})
+                    .get("full_refresh_no_slice_in_params", False)
+                ):
                     incremental_sync_component_definition["step"] = None
                     incremental_sync_component_definition["cursor_granularity"] = None
                     incremental_sync_component_definition["start_time_option"] = None

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -163,8 +163,12 @@ def update_with_cache_parent_configs(parent_configs: list[dict[str, Any]]) -> No
             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"]["use_cache"] = True
-                    parent_config["stream"]["retriever"]["incremental_retriever"]["requester"]["use_cache"] = True
+                    parent_config["stream"]["retriever"]["full_refresh_retriever"]["requester"][
+                        "use_cache"
+                    ] = True
+                    parent_config["stream"]["retriever"]["incremental_retriever"]["requester"][
+                        "use_cache"
+                    ] = True
                 else:
                     parent_config["stream"]["retriever"]["requester"]["use_cache"] = True
 
@@ -190,8 +194,12 @@ 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"]["use_cache"] = True
-                    stream_config["retriever"]["incremental_retriever"]["requester"]["use_cache"] = True
+                    stream_config["retriever"]["full_refresh_retriever"]["requester"][
+                        "use_cache"
+                    ] = True
+                    stream_config["retriever"]["incremental_retriever"]["requester"][
+                        "use_cache"
+                    ] = True
                 else:
                     stream_config["retriever"]["requester"]["use_cache"] = True
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1808,7 +1808,10 @@ def _build_incremental_cursor(
         stream_slicer: Optional[PartitionRouter],
         config: Config,
     ) -> Optional[StreamSlicer]:
-        if model.retriever.type == "StateDelegatingRetriever" and model.retriever.full_refresh_no_slice_in_params:
+        if (
+            model.retriever.type == "StateDelegatingRetriever"
+            and model.retriever.full_refresh_no_slice_in_params
+        ):
             model.incremental_sync.step = None
             model.incremental_sync.cursor_granularity = None
             model.incremental_sync.start_time_option = None
@@ -1918,7 +1921,11 @@ def _merge_stream_slicers(
                 raise ValueError("Per partition state is not supported yet for AsyncRetriever.")
 
         if retriever_model.type == "StateDelegatingRetriever":
-            retriever_model = retriever_model.incremental_retriever if self._connector_state_manager.get_stream_state(model.name, None) else retriever_model.full_refresh_retriever
+            retriever_model = (
+                retriever_model.incremental_retriever
+                if self._connector_state_manager.get_stream_state(model.name, None)
+                else retriever_model.full_refresh_retriever
+            )
 
         stream_slicer = self._build_stream_slicer_from_partition_router(retriever_model, config)
 
@@ -2746,8 +2753,11 @@ def create_state_delegating_retriever(
         client_side_incremental_sync: Optional[Dict[str, Any]] = None,
         transformations: List[RecordTransformation],
     ) -> SimpleRetriever:
-
-        retriever_model = model.incremental_retriever if self._connector_state_manager.get_stream_state(name, None) else model.full_refresh_retriever
+        retriever_model = (
+            model.incremental_retriever
+            if self._connector_state_manager.get_stream_state(name, None)
+            else model.full_refresh_retriever
+        )
 
         return self._create_component_from_model(
             model=retriever_model,