mypy

Author: maxime.c

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3480,13 +3480,15 @@ def create_state_delegating_stream(
                 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 = self._get_state_delegating_stream_model(has_parent_state, model)
+        stream_model = self._get_state_delegating_stream_model(
+            False if has_parent_state is None else has_parent_state, model
+        )
 
         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
 
     def _get_state_delegating_stream_model(
         self, has_parent_state: bool, model: StateDelegatingStreamModel
-    ):
+    ) -> DeclarativeStreamModel:
         return (
             model.incremental_stream
             if self._connector_state_manager.get_stream_state(model.name, None) or has_parent_state
@@ -3874,12 +3876,16 @@ def _instantiate_parent_stream_state_manager(
                 if not parent_state and not isinstance(parent_state, dict):
                     cursor_values = child_state.values()
                     if cursor_values:
-                        incremental_sync_model = (
-                            model.stream.incremental_sync
+                        incremental_sync_model: Union[
+                            DatetimeBasedCursorModel,
+                            IncrementingCountCursorModel,
+                            CustomIncrementalSyncModel,
+                        ] = (
+                            model.stream.incremental_sync  # type: ignore  # if we are there, it is because there is incremental_dependency and therefore there is an incremental_sync on the parent stream
                             if isinstance(model.stream, DeclarativeStreamModel)
                             else self._get_state_delegating_stream_model(
                                 has_parent_state, model.stream
-                            )
+                            ).incremental_sync
                         )
                         cursor_field = InterpolatedString.create(
                             incremental_sync_model.cursor_field,