fix: treat empty string cursor field as no cursor field (#875)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -4339,8 +4339,14 @@ def _get_catalog_defined_cursor_field(
         configured_stream = self._stream_name_to_configured_stream.get(stream_name)
 
         # Depending on the operation is being performed, there may not be a configured stream yet. In this
-        # case we return None which will then use the default cursor field defined on the cursor model
-        if not configured_stream or not configured_stream.cursor_field:
+        # case we return None which will then use the default cursor field defined on the cursor model.
+        # We also treat cursor_field: [""] (list with empty string) as no cursor field, since this can
+        # occur when the platform serializes "no cursor configured" streams incorrectly.
+        if (
+            not configured_stream
+            or not configured_stream.cursor_field
+            or not configured_stream.cursor_field[0]
+        ):
             return None
         elif len(configured_stream.cursor_field) > 1:
             raise ValueError(

airbyte_cdk/sources/streams/concurrent/helpers.py

@@ -34,7 +34,8 @@ def get_cursor_field_from_stream(stream: Stream) -> Optional[str]:
             raise ValueError(
                 f"Nested cursor fields are not supported. Got {stream.cursor_field} for {stream.name}"
             )
-        elif len(stream.cursor_field) == 0:
+        elif len(stream.cursor_field) == 0 or not stream.cursor_field[0]:
+            # Treat cursor_field: [""] (list with empty string) as no cursor field
             return None
         else:
             return stream.cursor_field[0]

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -5051,6 +5051,12 @@ def test_create_stream_with_multiple_schema_loaders():
             "updated_at",
             id="test_allow_catalog_defined_cursor_field_false_defaults_to_stream_defined_cursor_field",
         ),
+        pytest.param(
+            True,
+            "",
+            "updated_at",
+            id="test_empty_string_catalog_cursor_field_defaults_to_stream_defined_cursor_field",
+        ),
     ],
 )
 def test_catalog_defined_cursor_field(