Low-code CDK: safe get response.json (#18931)

* low-code cdk: safe get response.json

* flake fix

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.5.4
+Low-code: Get response.json in a safe way
+
 ## 0.5.3
 Low-code: Replace EmptySchemaLoader with DefaultSchemaLoader to retain backwards compatibility
 Low-code: Evaluate backoff strategies at runtime

airbyte-cdk/python/airbyte_cdk/sources/declarative/decoders/json_decoder.py

@@ -19,4 +19,7 @@ class JsonDecoder(Decoder, JsonSchemaMixin):
     options: InitVar[Mapping[str, Any]]
 
     def decode(self, response: requests.Response) -> Union[Mapping[str, Any], List]:
-        return response.json() or {}
+        try:
+            return response.json()
+        except requests.exceptions.JSONDecodeError:
+            return {}

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py

@@ -76,16 +76,23 @@ def _matches_filter(self, response: requests.Response) -> Optional[ResponseActio
         else:
             return None
 
+    @staticmethod
+    def _safe_response_json(response: requests.Response) -> dict:
+        try:
+            return response.json()
+        except requests.exceptions.JSONDecodeError:
+            return {}
+
     def _create_error_message(self, response: requests.Response) -> str:
         """
         Construct an error message based on the specified message template of the filter.
         :param response: The HTTP response which can be used during interpolation
         :return: The evaluated error message string to be emitted
         """
-        return self.error_message.eval(self.config, response=response.json(), headers=response.headers)
+        return self.error_message.eval(self.config, response=self._safe_response_json(response), headers=response.headers)
 
     def _response_matches_predicate(self, response: requests.Response) -> bool:
-        return self.predicate and self.predicate.eval(None, response=response.json(), headers=response.headers)
+        return self.predicate and self.predicate.eval(None, response=self._safe_response_json(response), headers=response.headers)
 
     def _response_contains_error_message(self, response: requests.Response) -> bool:
         if not self.error_message_contains:

airbyte-cdk/python/setup.py

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

airbyte-cdk/python/unit_tests/sources/declarative/decoders/test_json_decoder.py

@@ -0,0 +1,16 @@
+#
+# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+#
+
+import pytest
+import requests
+from airbyte_cdk.sources.declarative.decoders.json_decoder import JsonDecoder
+
+
+@pytest.mark.parametrize(
+    "response_body, expected_json", (("", {}), ('{"healthcheck": {"status": "ok"}}', {"healthcheck": {"status": "ok"}}))
+)
+def test_json_decoder(requests_mock, response_body, expected_json):
+    requests_mock.register_uri("GET", "https://airbyte.io/", text=response_body)
+    response = requests.get("https://airbyte.io/")
+    assert JsonDecoder(options={}).decode(response) == expected_json

airbyte-cdk/python/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py

@@ -2,9 +2,10 @@
 # Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 #
 
-from unittest.mock import MagicMock
+import json
 
 import pytest
+import requests
 from airbyte_cdk.sources.declarative.requesters.error_handlers import HttpResponseFilter
 from airbyte_cdk.sources.declarative.requesters.error_handlers.response_action import ResponseAction
 from airbyte_cdk.sources.declarative.requesters.error_handlers.response_status import ResponseStatus
@@ -86,18 +87,21 @@
             "",
             None,
             "",
-            {"status_code": 403, "headers": {"error", "authentication_error"}, "json": {"reason": "permission denied"}},
+            {"status_code": 403, "headers": {"error": "authentication_error"}, "json": {"reason": "permission denied"}},
             None,
             id="test_response_does_not_match_filter",
         ),
     ],
 )
-def test_matches(action, http_codes, predicate, error_contains, back_off, error_message, response, expected_response_status):
-    mock_response = MagicMock()
-    mock_response.status_code = response.get("status_code")
-    mock_response.headers = response.get("headers")
-    mock_response.json.return_value = response.get("json")
-
+def test_matches(requests_mock, action, http_codes, predicate, error_contains, back_off, error_message, response, expected_response_status):
+    requests_mock.register_uri(
+        "GET",
+        "https://airbyte.io/",
+        text=response.get("json") and json.dumps(response.get("json")),
+        headers=response.get("headers") or {},
+        status_code=response.get("status_code"),
+    )
+    response = requests.get("https://airbyte.io/")
     response_filter = HttpResponseFilter(
         action=action,
         config={},
@@ -108,7 +112,7 @@ def test_matches(action, http_codes, predicate, error_contains, back_off, error_
         error_message=error_message,
     )
 
-    actual_response_status = response_filter.matches(mock_response, backoff_time=back_off or 10)
+    actual_response_status = response_filter.matches(response, backoff_time=back_off or 10)
     if expected_response_status:
         assert actual_response_status.action == expected_response_status.action
         assert actual_response_status.retry_in == expected_response_status.retry_in