🐛 Source SalesForce: better detect API type (#16086)

Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>

Author: grubberr

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

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

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

@@ -8654,7 +8654,7 @@
  supportsNormalization: false
  supportsDBT: false
  supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-salesforce:1.0.14"
+- dockerImage: "airbyte/source-salesforce:1.0.15"
  spec:
  documentationUrl: "https://docs.airbyte.com/integrations/sources/salesforce"
  connectionSpecification:

airbyte-integrations/connectors/source-salesforce/Dockerfile

@@ -13,5 +13,5 @@ RUN pip install .
 
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=1.0.14
+LABEL io.airbyte.version=1.0.15
 LABEL io.airbyte.name=airbyte/source-salesforce

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

@@ -3,10 +3,10 @@
 #
 
 import concurrent.futures
+import logging
 from typing import Any, List, Mapping, Optional, Tuple
 
 import requests # type: ignore[import]
-from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.models import ConfiguredAirbyteCatalog
 from requests import adapters as request_adapters
 from requests.exceptions import HTTPError, RequestException # type: ignore[import]
@@ -178,9 +178,12 @@
 
 
 class Salesforce:
- logger = AirbyteLogger()
+ logger = logging.getLogger("airbyte")
  version = "v52.0"
  parallel_tasks_size = 100
+ # https://developer.salesforce.com/docs/atlas.en-us.salesforce_app_limits_cheatsheet.meta/salesforce_app_limits_cheatsheet/salesforce_app_limits_platform_api.htm
+ # Request Size Limits
+ REQUEST_SIZE_LIMITS = 16_384
 
  def __init__(
  self,

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

@@ -36,6 +36,26 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
 
  return True, None
 
+ @classmethod
+ def _get_api_type(cls, stream_name, properties, stream_state):
+ # 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 properties.items() if value.get("format") == "base64" or "object" in value["type"]
+ }
+ properties_length = len(",".join(p for p in properties))
+
+ rest_required = stream_name in UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS or properties_not_supported_by_bulk
+ # If we have a lot of properties we can overcome REST API URL length and get an error: "reason: URI Too Long".
+ # For such cases connector tries to use BULK API because it uses POST request and passes properties in the request body.
+ bulk_required = properties_length + 2000 > Salesforce.REQUEST_SIZE_LIMITS
+
+ if bulk_required and not rest_required:
+ return "bulk"
+ elif rest_required and not bulk_required:
+ return "rest"
+ elif not bulk_required and not rest_required:
+ return "rest" if stream_state else "bulk"
+
  @classmethod
  def generate_streams(
  cls,
@@ -51,19 +71,15 @@ def generate_streams(
  for stream_name, sobject_options in stream_objects.items():
  streams_kwargs = {"sobject_options": sobject_options}
  stream_state = state.get(stream_name, {}) if state else {}
-
  selected_properties = stream_properties.get(stream_name, {}).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"]
- }
 
- if stream_state or stream_name in UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS or properties_not_supported_by_bulk:
-  # Use REST API
+ api_type = cls._get_api_type(stream_name, selected_properties, stream_state)
+ if api_type == "rest":
  full_refresh, incremental = SalesforceStream, IncrementalSalesforceStream
- else:
- # Use BULK API
+ elif api_type == "bulk":
  full_refresh, incremental = BulkSalesforceStream, BulkIncrementalSalesforceStream
+ else:
+ raise Exception(f"Stream {stream_name} cannot be processed by REST or BULK API.")
 
  json_schema = stream_properties.get(stream_name, {})
  pk, replication_key = sf_object.get_pk_and_replication_key(json_schema)

docs/integrations/sources/salesforce.md

@@ -119,6 +119,7 @@ Now that you have set up the Salesforce source connector, check out the followin
 
 | Version | Date | Pull Request | Subject |
 |:--------|:-----------|:-------------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------|
+| 1.0.15 | 2022-08-30 | [16086](https://github.com/airbytehq/airbyte/pull/16086) | Improve API type detection |
 | 1.0.14 | 2022-08-29 | [16119](https://github.com/airbytehq/airbyte/pull/16119) | Exclude `KnowledgeArticleVersion` from using bulk API |
 | 1.0.13 | 2022-08-23 | [15901](https://github.com/airbytehq/airbyte/pull/15901) | Exclude `KnowledgeArticle` from using bulk API |
 | 1.0.12 | 2022-08-09 | [15444](https://github.com/airbytehq/airbyte/pull/15444) | Fixed bug when `Bulk Job` was timeout by the connector, but remained running on the server |