format

Author: maxime.c

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2170,7 +2170,7 @@ def _build_incremental_cursor(
         )
 
         if model.incremental_sync and (
-                stream_slicer and not isinstance(stream_slicer, SinglePartitionRouter)
+            stream_slicer and not isinstance(stream_slicer, SinglePartitionRouter)
         ):
             if model.retriever.type == "AsyncRetriever":
                 stream_name = model.name or ""
@@ -2244,7 +2244,11 @@ def _build_concurrent_cursor(
         else:
             state_transformations = []
 
-        if model.incremental_sync and stream_slicer and not isinstance(stream_slicer, SinglePartitionRouter):
+        if (
+            model.incremental_sync
+            and stream_slicer
+            and not isinstance(stream_slicer, SinglePartitionRouter)
+        ):
             return self.create_concurrent_cursor_from_perpartition_cursor(  # type: ignore # This is a known issue that we are creating and returning a ConcurrentCursor which does not technically implement the (low-code) StreamSlicer. However, (low-code) StreamSlicer and ConcurrentCursor both implement StreamSlicer.stream_slices() which is the primary method needed for checkpointing
                 state_manager=self._connector_state_manager,
                 model_type=DatetimeBasedCursorModel,

unit_tests/connector_builder/test_connector_builder_handler.py

@@ -785,7 +785,11 @@ def test_config_update() -> None:
         "refresh_token": "a refresh token",
     }
     source = ConcurrentDeclarativeSource(
-        catalog=None, config=config, state=None, source_config=manifest, emit_connector_builder_messages=True
+        catalog=None,
+        config=config,
+        state=None,
+        source_config=manifest,
+        emit_connector_builder_messages=True,
     )
 
     refresh_request_response = {