Fix mypy

Author: lazebnyi

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -211,8 +211,11 @@ def _group_streams(
                     stream_name=declarative_stream.name, namespace=declarative_stream.namespace
                 )
 
-                name_to_stream_mapping[declarative_stream.name] = name_to_stream_mapping[declarative_stream.name]["incremental_stream"] \
-                    if stream_state else name_to_stream_mapping[declarative_stream.name]["full_refresh_stream"]
+                name_to_stream_mapping[declarative_stream.name] = (
+                    name_to_stream_mapping[declarative_stream.name]["incremental_stream"]
+                    if stream_state
+                    else name_to_stream_mapping[declarative_stream.name]["full_refresh_stream"]
+                )
 
             if isinstance(declarative_stream, DeclarativeStream) and (
                 name_to_stream_mapping[declarative_stream.name]["retriever"]["type"]

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -29,10 +29,10 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DeclarativeStream as DeclarativeStreamModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import Spec as SpecModel
 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,
 )
@@ -146,7 +146,9 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
 
         source_streams = [
             self._constructor.create_component(
-                StateDelegatingStreamModel if stream_config.get("type") == StateDelegatingStreamModel.__name__ else DeclarativeStreamModel,
+                StateDelegatingStreamModel
+                if stream_config.get("type") == StateDelegatingStreamModel.__name__
+                else DeclarativeStreamModel,
                 stream_config,
                 config,
                 emit_connector_builder_messages=self._emit_connector_builder_messages,
@@ -197,12 +199,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["type"] == "StateDelegatingStream":
-                    stream_config["full_refresh_stream"]["retriever"]["requester"][
-                        "use_cache"
-                    ] = True
-                    stream_config["incremental_stream"]["retriever"]["requester"][
-                        "use_cache"
-                    ] = True
+                    stream_config["full_refresh_stream"]["retriever"]["requester"]["use_cache"] = (
+                        True
+                    )
+                    stream_config["incremental_stream"]["retriever"]["requester"]["use_cache"] = (
+                        True
+                    )
                 else:
                     stream_config["retriever"]["requester"]["use_cache"] = True
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -204,9 +204,6 @@
 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 (
     DefaultErrorHandler as DefaultErrorHandlerModel,
 )
@@ -355,6 +352,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 (
+    StateDelegatingStream as StateDelegatingStreamModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     StreamConfig as StreamConfigModel,
 )
@@ -1793,13 +1793,17 @@ def _build_stream_slicer_from_partition_router(
             if isinstance(stream_slicer_model, list):
                 return CartesianProductStreamSlicer(
                     [
-                        self._create_component_from_model(model=slicer, config=config, stream_name=stream_name or "")
+                        self._create_component_from_model(
+                            model=slicer, config=config, stream_name=stream_name or ""
+                        )
                         for slicer in stream_slicer_model
                     ],
                     parameters={},
                 )
             else:
-                return self._create_component_from_model(model=stream_slicer_model, config=config, stream_name=stream_name or "")  # type: ignore[no-any-return] # Will be created PartitionRouter as stream_slicer_model is model.partition_router
+                return self._create_component_from_model(
+                    model=stream_slicer_model, config=config, stream_name=stream_name or ""
+                )  # type: ignore[no-any-return] # Will be created PartitionRouter as stream_slicer_model is model.partition_router
         return None
 
     def _build_incremental_cursor(
@@ -2478,7 +2482,9 @@ def create_page_increment(
     def create_parent_stream_config(
         self, model: ParentStreamConfigModel, config: Config, **kwargs: Any
     ) -> ParentStreamConfig:
-        declarative_stream = self._create_component_from_model(model.stream, config=config, **kwargs)
+        declarative_stream = self._create_component_from_model(
+            model.stream, config=config, **kwargs
+        )
         request_option = (
             self._create_component_from_model(model.request_option, config=config)
             if model.request_option
@@ -2719,12 +2725,26 @@ def create_simple_retriever(
             parameters=model.parameters or {},
         )
 
-    def create_state_delegating_stream(self, model: StateDelegatingStreamModel, config: Config, child_state: Optional[MutableMapping[str, Any]] = None, **kwargs: Any
+    def create_state_delegating_stream(
+        self,
+        model: StateDelegatingStreamModel,
+        config: Config,
+        child_state: Optional[MutableMapping[str, Any]] = None,
+        **kwargs: Any,
     ) -> DeclarativeStream:
-        if model.full_refresh_stream.name != model.name or model.name != model.incremental_stream.name:
-            raise ValueError(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}.")
+        if (
+            model.full_refresh_stream.name != model.name
+            or model.name != model.incremental_stream.name
+        ):
+            raise ValueError(
+                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 = model.incremental_stream if self._connector_state_manager.get_stream_state(model.name, None) or child_state else model.full_refresh_stream
+        stream_model = (
+            model.incremental_stream
+            if self._connector_state_manager.get_stream_state(model.name, None) or child_state
+            else model.full_refresh_stream
+        )
 
         return self._create_component_from_model(stream_model, config=config, **kwargs)  # type: ignore[no-any-return]  # Will be created DeclarativeStream as stream_model is stream description
 
@@ -2981,8 +3001,14 @@ def _create_message_repository_substream_wrapper(
                 self._evaluate_log_level(self._emit_connector_builder_messages),
             ),
         )
-        child_state = self._connector_state_manager.get_stream_state(kwargs.get("stream_name", ""), None) if model.incremental_dependency or False else None
-        return substream_factory._create_component_from_model(model=model, config=config, child_state=child_state, **kwargs)
+        child_state = (
+            self._connector_state_manager.get_stream_state(kwargs.get("stream_name", ""), None)
+            if model.incremental_dependency or False
+            else None
+        )
+        return substream_factory._create_component_from_model(
+            model=model, config=config, child_state=child_state, **kwargs
+        )
 
     @staticmethod
     def create_wait_time_from_header(

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -115,7 +115,7 @@
                         "inject_into": "request_parameter",
                     },
                 },
-            }
+            },
         },
     },
     "streams": [{"$ref": "#/definitions/TestStream"}],