fix cdk bug to send legacy format if connector overrides read() (#16566)

* fix cdk bug to send legacy format if connector overrides read()

* fix comment

* update changelog and setup.py

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.1.83
+- Fix per-stream to send legacy format for connectors that override read
+
 ## 0.1.82
 - Freeze dataclasses-jsonschema to 2.15.1
 

airbyte-cdk/python/airbyte_cdk/sources/source.py

@@ -66,7 +66,11 @@ def read_state(self, state_path: str) -> List[AirbyteStateMessage]:
                     parsed_state_messages.append(parsed_message)
                 return parsed_state_messages
             else:
-                # When the legacy JSON object format is received, always outputting an AirbyteStateMessage simplifies processing downstream
+                # Existing connectors that override read() might not be able to interpret the new state format. We temporarily
+                # send state in the old format for these connectors, but once all have been upgraded, this block can be removed
+                # vars(self.__class__) checks if the current class directly overrides the read() function
+                if "read" in vars(self.__class__):
+                    return state_obj
                 return [AirbyteStateMessage(type=AirbyteStateType.LEGACY, data=state_obj)]
         return []
 

airbyte-cdk/python/setup.py

@@ -15,7 +15,7 @@
 
 setup(
     name="airbyte-cdk",
-    version="0.1.82",
+    version="0.1.83",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",

airbyte-cdk/python/unit_tests/sources/test_source.py

@@ -6,7 +6,7 @@
 import logging
 import tempfile
 from contextlib import nullcontext as does_not_raise
-from typing import Any, Mapping, MutableMapping
+from typing import Any, List, Mapping, MutableMapping, Optional, Tuple
 from unittest.mock import MagicMock
 
 import pytest
@@ -41,6 +41,14 @@ def discover(self, logger: logging.Logger, config: Mapping[str, Any]):
         pass
 
 
+class MockAbstractSource(AbstractSource):
+    def check_connection(self, *args, **kwargs) -> Tuple[bool, Optional[Any]]:
+        return True, ""
+
+    def streams(self, *args, **kwargs) -> List[Stream]:
+        return []
+
+
 @pytest.fixture
 def source():
     return MockSource()
@@ -212,11 +220,7 @@ def streams(self, config):
         ),
         pytest.param(
             {"movies": {"created_at": "2009-07-19"}, "directors": {"id": "villeneuve_denis"}},
-            [
-                AirbyteStateMessage(
-                    type=AirbyteStateType.LEGACY, data={"movies": {"created_at": "2009-07-19"}, "directors": {"id": "villeneuve_denis"}}
-                )
-            ],
+            {"movies": {"created_at": "2009-07-19"}, "directors": {"id": "villeneuve_denis"}},
             does_not_raise(),
             id="test_incoming_legacy_state",
         ),
@@ -283,6 +287,20 @@ def test_read_state(source, incoming_state, expected_state, expected_error):
             assert actual == expected_state
 
 
+def test_read_state_sends_new_legacy_format_if_source_does_not_implement_read():
+    expected_state = [
+        AirbyteStateMessage(
+            type=AirbyteStateType.LEGACY, data={"movies": {"created_at": "2009-07-19"}, "directors": {"id": "villeneuve_denis"}}
+        )
+    ]
+    source = MockAbstractSource()
+    with tempfile.NamedTemporaryFile("w") as state_file:
+        state_file.write(json.dumps({"movies": {"created_at": "2009-07-19"}, "directors": {"id": "villeneuve_denis"}}))
+        state_file.flush()
+        actual = source.read_state(state_file.name)
+        assert actual == expected_state
+
+
 def test_read_state_nonexistent(source):
     assert [] == source.read_state("")