🐛 Source Salesforce: Fix error during generating schema (#9478)

* Add filtering by queryable flag for available streams

* Updated PR number and version

* add logs for failed bulk jobs

* Added logs for collect stream objects options if code 404

* Fixed unittests

* Fixed integration tests

* Bumped docker version

* Update version in seed

* Update streams.py

* Added logger for print not queriable streams

Co-authored-by: antixar <antixar@gmail.com>

Author: lazebnyi

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b117307c-14b6-41aa-9422-947e34922962.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "b117307c-14b6-41aa-9422-947e34922962",
   "name": "Salesforce",
   "dockerRepository": "airbyte/source-salesforce",
-  "dockerImageTag": "0.1.13",
+  "dockerImageTag": "0.1.18",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/salesforce",
   "icon": "salesforce.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -634,7 +634,7 @@
 - name: Salesforce
   sourceDefinitionId: b117307c-14b6-41aa-9422-947e34922962
   dockerRepository: airbyte/source-salesforce
-  dockerImageTag: 0.1.17
+  dockerImageTag: 0.1.18
   documentationUrl: https://docs.airbyte.io/integrations/sources/salesforce
   icon: salesforce.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -6700,7 +6700,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-salesforce:0.1.17"
+- dockerImage: "airbyte/source-salesforce:0.1.18"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/salesforce"
     connectionSpecification:

airbyte-integrations/connectors/source-salesforce/Dockerfile

@@ -25,5 +25,5 @@ COPY source_salesforce ./source_salesforce
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.17
+LABEL io.airbyte.version=0.1.18
 LABEL io.airbyte.name=airbyte/source-salesforce

airbyte-integrations/connectors/source-salesforce/source_salesforce/api.py

@@ -208,9 +208,16 @@ def filter_streams(self, stream_name: str) -> bool:
 
     def get_validated_streams(self, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog = None):
         salesforce_objects = self.describe()["sobjects"]
-        stream_names = [stream_object["name"] for stream_object in salesforce_objects]
+        stream_objects = []
+        for stream_object in salesforce_objects:
+            if stream_object["queryable"]:
+                stream_objects.append(stream_object)
+            else:
+                self.logger.warn(f"Stream {stream_object['name']} is not queryable and will be ignored.")
+
+        stream_names = [stream_object["name"] for stream_object in stream_objects]
         if catalog:
-            return [configured_stream.stream.name for configured_stream in catalog.streams]
+            return [configured_stream.stream.name for configured_stream in catalog.streams], stream_objects
 
         if config.get("streams_criteria"):
             filtered_stream_list = []
@@ -221,7 +228,8 @@ def get_validated_streams(self, config: Mapping[str, Any], catalog: ConfiguredAi
             stream_names = list(set(filtered_stream_list))
 
         validated_streams = [stream_name for stream_name in stream_names if self.filter_streams(stream_name)]
-        return validated_streams
+        validated_stream_objects = [stream_object for stream_object in stream_objects if stream_object["name"] in validated_streams]
+        return validated_streams, validated_stream_objects
 
     @default_backoff_handler(max_tries=5, factor=15)
     def _make_request(
@@ -253,18 +261,20 @@ def login(self):
         self.access_token = auth["access_token"]
         self.instance_url = auth["instance_url"]
 
-    def describe(self, sobject: str = None) -> Mapping[str, Any]:
+    def describe(self, sobject: str = None, stream_objects: List = None) -> Mapping[str, Any]:
         """Describes all objects or a specific object"""
         headers = self._get_standard_headers()
 
         endpoint = "sobjects" if not sobject else f"sobjects/{sobject}/describe"
 
         url = f"{self.instance_url}/services/data/{self.version}/{endpoint}"
         resp = self._make_request("GET", url, headers=headers)
+        if resp.status_code == 404:
+            self.logger.error(f"Filtered stream objects: {stream_objects}")
         return resp.json()
 
-    def generate_schema(self, stream_name: str = None) -> Mapping[str, Any]:
-        response = self.describe(stream_name)
+    def generate_schema(self, stream_name: str = None, stream_objects: List = None) -> Mapping[str, Any]:
+        response = self.describe(stream_name, stream_objects)
         schema = {"$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "additionalProperties": True, "properties": {}}
         for field in response["fields"]:
             schema["properties"][field["name"]] = self.field_to_property_schema(field)

airbyte-integrations/connectors/source-salesforce/source_salesforce/source.py

@@ -29,7 +29,7 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
 
     @classmethod
     def generate_streams(
-        cls, config: Mapping[str, Any], stream_names: List[str], sf_object: Salesforce, state: Mapping[str, Any] = None
+        cls, config: Mapping[str, Any], stream_names: List[str], sf_object: Salesforce, state: Mapping[str, Any] = None, stream_objects: List = None
     ) -> List[Stream]:
         """ "Generates a list of stream by their names. It can be used for different tests too"""
         authenticator = TokenAuthenticator(sf_object.access_token)
@@ -38,7 +38,7 @@ def generate_streams(
             streams_kwargs = {}
             stream_state = state.get(stream_name, {}) if state else {}
 
-            selected_properties = sf_object.generate_schema(stream_name).get("properties", {})
+            selected_properties = sf_object.generate_schema(stream_name, stream_objects).get("properties", {})
             # Salesforce BULK API currently does not support loading fields with data type base64 and compound data
             properties_not_supported_by_bulk = {
                 key: value for key, value in selected_properties.items() if value.get("format") == "base64" or "object" in value["type"]
@@ -52,7 +52,7 @@ def generate_streams(
                 full_refresh, incremental = BulkSalesforceStream, BulkIncrementalSalesforceStream
                 streams_kwargs["wait_timeout"] = config.get("wait_timeout")
 
-            json_schema = sf_object.generate_schema(stream_name)
+            json_schema = sf_object.generate_schema(stream_name, stream_objects)
             pk, replication_key = sf_object.get_pk_and_replication_key(json_schema)
             streams_kwargs.update(dict(sf_api=sf_object, pk=pk, stream_name=stream_name, schema=json_schema, authenticator=authenticator))
             if replication_key and stream_name not in UNSUPPORTED_FILTERING_STREAMS:
@@ -64,8 +64,8 @@ def generate_streams(
 
     def streams(self, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog = None, state: Mapping[str, Any] = None) -> List[Stream]:
         sf = self._get_sf_object(config)
-        stream_names = sf.get_validated_streams(config=config, catalog=catalog)
-        return self.generate_streams(config, stream_names, sf, state=state)
+        stream_names, stream_objects = sf.get_validated_streams(config=config, catalog=catalog)
+        return self.generate_streams(config, stream_names, sf, state=state, stream_objects=stream_objects)
 
     def read(
         self, logger: AirbyteLogger, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog, state: MutableMapping[str, Any] = None

airbyte-integrations/connectors/source-salesforce/source_salesforce/streams.py

@@ -133,6 +133,8 @@ def transform_empty_string_to_none(instance, schema):
     def _send_http_request(self, method: str, url: str, json: dict = None):
         headers = self.authenticator.get_auth_header()
         response = self._session.request(method, url=url, headers=headers, json=json)
+        if response.status_code not in [200, 204]:
+            self.logger.error(f"error body: {response.text}")
         response.raise_for_status()
         return response
 
@@ -141,6 +143,7 @@ def create_stream_job(self, query: str, url: str) -> Optional[str]:
         docs: https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/create_job.htm
         """
         json = {"operation": "queryAll", "query": query, "contentType": "CSV", "columnDelimiter": "COMMA", "lineEnding": "LF"}
+
         try:
             response = self._send_http_request("POST", url, json=json)
             job_id = response.json()["id"]
@@ -191,6 +194,10 @@ def wait_for_job(self, url: str) -> str:
             job_info = self._send_http_request("GET", url=url).json()
             job_status = job_info["state"]
             if job_status in ["JobComplete", "Aborted", "Failed"]:
+                if job_status != "JobComplete":
+                    # this is only job metadata without payload
+                    self.logger.error(f"JobStatus: {job_status}, full job response: {job_info}")
+
                 return job_status
 
             if delay_timeout < self.MAX_CHECK_INTERVAL_SECONDS:

airbyte-integrations/connectors/source-salesforce/unit_tests/unit_test.py

@@ -302,21 +302,38 @@ def test_discover_with_streams_criteria_param(streams_criteria, predicted_filter
     sf_object.describe = Mock(
         return_value={
             "sobjects": [
-                {"name": "Account"},
-                {"name": "AIApplications"},
-                {"name": "Leads"},
-                {"name": "LeadHistory"},
-                {"name": "Orders"},
-                {"name": "OrderHistory"},
-                {"name": "CustomStream"},
-                {"name": "CustomStreamHistory"},
+                {"name": "Account", "queryable": True},
+                {"name": "AIApplications", "queryable": True},
+                {"name": "Leads", "queryable": True},
+                {"name": "LeadHistory", "queryable": True},
+                {"name": "Orders", "queryable": True},
+                {"name": "OrderHistory", "queryable": True},
+                {"name": "CustomStream", "queryable": True},
+                {"name": "CustomStreamHistory", "queryable": True},
             ]
         }
     )
-    filtered_streams = sf_object.get_validated_streams(config=updated_config)
+    filtered_streams, _ = sf_object.get_validated_streams(config=updated_config)
     assert sorted(filtered_streams) == sorted(predicted_filtered_streams)
 
 
+def test_discover_only_queryable(stream_config):
+    sf_object = Salesforce(**stream_config)
+    sf_object.login = Mock()
+    sf_object.access_token = Mock()
+    sf_object.instance_url = "https://fase-account.salesforce.com"
+    sf_object.describe = Mock(
+        return_value={
+            "sobjects": [
+                {"name": "Account", "queryable": True},
+                {"name": "Leads", "queryable": False},
+            ]
+        }
+    )
+    filtered_streams, _ = sf_object.get_validated_streams(config=stream_config)
+    assert filtered_streams == ["Account"]
+
+
 def test_pagination_rest(stream_config, stream_api):
     stream_name = "ActiveFeatureLicenseMetric"
     state = {stream_name: {"SystemModstamp": "2122-08-22T05:08:29.000Z"}}

docs/integrations/sources/salesforce.md

@@ -737,6 +737,7 @@ List of available streams:
 
 | Version | Date       | Pull Request | Subject                                                                   |
 |:--------|:-----------| :--- |:--------------------------------------------------------------------------|
+| 0.1.18  | 2022-01-20 | [9478](https://github.com/airbytehq/airbyte/pull/9478) | Add available stream filtering by `queryable` flag |
 | 0.1.17  | 2022-01-19 | [9302](https://github.com/airbytehq/airbyte/pull/9302) | Deprecate API Type parameter                                             |
 | 0.1.16  | 2022-01-18 | [9151](https://github.com/airbytehq/airbyte/pull/9151) | Fix pagination in REST API streams                                       |
 | 0.1.15  | 2022-01-11 | [9409](https://github.com/airbytehq/airbyte/pull/9409) | Correcting the presence of an extra `else` handler in the error handling |