🐛 Salesforce Connector: handling 400 error, while sobject doesn't sup…

…port query or queryAll requests (airbytehq#9386)

* Salesforce Connector: handling 400 error, while sobject doesn't support query or queryAll requests

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

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

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

@@ -6310,7 +6310,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-salesforce:0.1.13"
+- dockerImage: "airbyte/source-salesforce:0.1.14"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/salesforce"
     connectionSpecification:
@@ -6324,6 +6324,9 @@
       - "api_type"
       additionalProperties: false
       properties:
+        auth_type:
+          type: "string"
+          const: "Client"
         client_id:
           title: "Client ID"
           description: "The Consumer Key that can be found when viewing your app in\
@@ -6409,15 +6412,48 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-    authSpecification:
-      auth_type: "oauth2.0"
-      oauth2Specification:
-        rootObject: []
-        oauthFlowInitParameters:
-        - - "client_id"
-        - - "client_secret"
-        oauthFlowOutputParameters:
-        - - "refresh_token"
+    advanced_auth:
+      auth_flow_type: "oauth2.0"
+      predicate_key:
+      - "auth_type"
+      predicate_value: "Client"
+      oauth_config_specification:
+        oauth_user_input_from_connector_config_specification:
+          type: "object"
+          additionalProperties: false
+          properties:
+            is_sandbox:
+              type: "boolean"
+              path_in_connector_config:
+              - "is_sandbox"
+        complete_oauth_output_specification:
+          type: "object"
+          additionalProperties: false
+          properties:
+            refresh_token:
+              type: "string"
+              path_in_connector_config:
+              - "refresh_token"
+        complete_oauth_server_input_specification:
+          type: "object"
+          additionalProperties: false
+          properties:
+            client_id:
+              type: "string"
+            client_secret:
+              type: "string"
+        complete_oauth_server_output_specification:
+          type: "object"
+          additionalProperties: false
+          properties:
+            client_id:
+              type: "string"
+              path_in_connector_config:
+              - "client_id"
+            client_secret:
+              type: "string"
+              path_in_connector_config:
+              - "client_secret"
 - dockerImage: "airbyte/source-search-metrics:0.1.1"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/seacrh-metrics"

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.13
+LABEL io.airbyte.version=0.1.14
 LABEL io.airbyte.name=airbyte/source-salesforce

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

@@ -90,9 +90,20 @@ def read_records(self, **kwargs) -> Iterable[Mapping[str, Any]]:
         try:
             yield from super().read_records(**kwargs)
         except exceptions.HTTPError as error:
+            """
+            There are several types of Salesforce sobjects that require additional processing:
+              1. Sobjects for which the user, after setting up the data using Airbyte, restricted access,
+                 and we will receive 403 HTTP errors.
+              2. There are streams that do not allow you to make a sample using Salesforce `query` or `queryAll`.
+                 And since we use a dynamic method of generating streams for Salesforce connector - at the stage of discover,
+                 we cannot filter out these streams, so we catch them at the stage of reading data.
+            """
             error_data = error.response.json()[0]
-            if error.response.status_code == codes.FORBIDDEN and not error_data.get("errorCode", "") == "REQUEST_LIMIT_EXCEEDED":
-                self.logger.error(f"Cannot receive data for stream '{self.name}', error message: '{error_data.get('message')}'")
+            if error.response.status_code in [codes.FORBIDDEN, codes.BAD_REQUEST]:
+                error_code = error_data.get("errorCode", "")
+                if error_code != "REQUEST_LIMIT_EXCEEDED" or error_code == "INVALID_TYPE_FOR_OPERATION":
+                    self.logger.error(f"Cannot receive data for stream '{self.name}', error message: '{error_data.get('message')}'")
+                    return
             else:
                 raise error
 

docs/integrations/sources/salesforce.md

@@ -737,6 +737,7 @@ List of available streams:
 
 | Version | Date       | Pull Request | Subject                                                                   |
 |:--------|:-----------| :--- |:--------------------------------------------------------------------------|
+| 0.1.14  | 2022-01-11 | [9386](https://github.com/airbytehq/airbyte/pull/9386) | Handling 400 error, while `sobject` doesn't support `query` or `queryAll` requests |
 | 0.1.13  | 2022-01-11 | [8797](https://github.com/airbytehq/airbyte/pull/8797) | Switched from authSpecification to advanced_auth in specefication         |
 | 0.1.12  | 2021-12-23 | [8871](https://github.com/airbytehq/airbyte/pull/8871) | Fix `examples` for new field in specification                             |
 | 0.1.11  | 2021-12-23 | [8871](https://github.com/airbytehq/airbyte/pull/8871) | Add the ability to filter streams by user                                 |