airbytehq · artem1205 · Aug 14, 2023 · Jul 12, 2023 · Jul 18, 2023 · Jul 18, 2023
@@ -1,38 +1,17 @@
-# Using alpine to remove several vulnerabilities frm slim image
-# https://security-tracker.debian.org/tracker/CVE-2023-29383
-# https://security-tracker.debian.org/tracker/CVE-2023-31484
-# https://security-tracker.debian.org/tracker/CVE-2016-2781
-FROM python:3.9-alpine3.18
-
-
-RUN apk add --update --no-cache \
-    build-base \
-    openssl-dev \
-    libffi-dev \
-    zlib-dev \
-    bzip2-dev
+FROM python:3.9-slim
 
+# Bash is installed for more convenient debugging.
+RUN apt-get update && apt-get install -y bash && rm -rf /var/lib/apt/lists/*
 
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 
 WORKDIR /airbyte/integration_code
 COPY source_salesforce ./source_salesforce
 COPY setup.py ./
 COPY main.py ./
-
-# Fixing https://nvd.nist.gov/vuln/detail/CVE-2022-40897
-# calling this twice as one upgrades the system pip /usr/local/bin/pip the
-# seconf time upgrades the under for the venv /opt/.venv/bin/pip
-RUN pip install --upgrade pip setuptools wheel && \
-    pip install .
-RUN pip install --upgrade pip setuptools
-
-# add default timezone settings
-ENV TZ UTC
-RUN cp /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
-
+RUN pip install .
 
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=2.1.1
+LABEL io.airbyte.version=2.1.2
 LABEL io.airbyte.name=airbyte/source-salesforce
@@ -5,15 +5,14 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: b117307c-14b6-41aa-9422-947e34922962
-  dockerImageTag: 2.1.1
+  dockerImageTag: 2.1.2
   dockerRepository: airbyte/source-salesforce
   githubIssueLabel: source-salesforce
   icon: salesforce.svg
   license: ELv2
   name: Salesforce
   registries:
     cloud:
-      dockerImageTag: 2.0.9
       enabled: true
     oss:
       enabled: true

@@ -5,9 +5,9 @@
 
 from setuptools import find_packages, setup
 
-MAIN_REQUIREMENTS = ["airbyte-cdk~=0.2", "vcrpy==4.1.1", "pandas"]
+MAIN_REQUIREMENTS = ["airbyte-cdk~=0.46", "pandas"]
 
-TEST_REQUIREMENTS = ["pytest~=6.1", "pytest-mock~=3.6", "requests_mock", "connector-acceptance-test", "pytest-timeout"]
+TEST_REQUIREMENTS = ["freezegun", "pytest~=6.1", "pytest-mock~=3.6", "requests_mock", "connector-acceptance-test", "pytest-timeout"]
 
 setup(
     name="source_salesforce",

@@ -296,7 +296,7 @@ def _make_request(
                 resp = self.session.post(url, headers=headers, data=body)
             resp.raise_for_status()
         except HTTPError as err:
-            self.logger.warn(f"http error body: {err.response.text}")
+            self.logger.warning(f"http error body: {err.response.text}")
             raise
         return resp
 

@@ -2,11 +2,10 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-
+import logging
 import sys
 
 import backoff
-from airbyte_cdk.logger import AirbyteLogger
 from airbyte_cdk.sources.streams.http.exceptions import DefaultBackoffException
 from requests import codes, exceptions  # type: ignore[import]
 
@@ -18,7 +17,7 @@
     exceptions.HTTPError,
 )
 
-logger = AirbyteLogger()
+logger = logging.getLogger("airbyte")
 
 
 def default_backoff_handler(max_tries: int, factor: int, **kwargs):

@@ -12,7 +12,7 @@
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
 from airbyte_cdk.sources.streams import Stream
-from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
+from airbyte_cdk.sources.streams.http.requests_native_auth import TokenAuthenticator
 from airbyte_cdk.sources.utils.schema_helpers import InternalConfig
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 from dateutil.relativedelta import relativedelta
@@ -21,6 +21,8 @@
 from .api import UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS, UNSUPPORTED_FILTERING_STREAMS, Salesforce
 from .streams import BulkIncrementalSalesforceStream, BulkSalesforceStream, Describe, IncrementalRestSalesforceStream, RestSalesforceStream
 
+logger = logging.getLogger("airbyte")
+
 
 class AirbyteStopSync(AirbyteTracedException):
     pass
@@ -59,13 +61,21 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
         return True, None
 
     @classmethod
-    def _get_api_type(cls, stream_name, properties):
+    def _get_api_type(cls, stream_name: str, properties: Mapping[str, Any], force_use_bulk_api: bool) -> str:
         # 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"]
         }
-        rest_required = stream_name in UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS or properties_not_supported_by_bulk
-        if rest_required:
+        rest_only = stream_name in UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS
+        if rest_only:
+            logger.warning(f"BULK API is not supported for stream: {stream_name}")
+            return "rest"
+        if force_use_bulk_api and properties_not_supported_by_bulk:
+            logger.warning(
+                f"Following properties will be excluded from stream: {stream_name} due to BULK API limitations: {list(properties_not_supported_by_bulk)}"
+            )
+            return "bulk"
+        if properties_not_supported_by_bulk:
             return "rest"
         return "bulk"
 
@@ -77,15 +87,14 @@ def generate_streams(
         sf_object: Salesforce,
     ) -> List[Stream]:
         """ "Generates a list of stream by their names. It can be used for different tests too"""
-        logger = logging.getLogger()
         authenticator = TokenAuthenticator(sf_object.access_token)
         stream_properties = sf_object.generate_schemas(stream_objects)
         streams = []
         for stream_name, sobject_options in stream_objects.items():
             streams_kwargs = {"sobject_options": sobject_options}
             selected_properties = stream_properties.get(stream_name, {}).get("properties", {})
 
-            api_type = cls._get_api_type(stream_name, selected_properties)
+            api_type = cls._get_api_type(stream_name, selected_properties, config.get("force_use_bulk_api", False))
             if api_type == "rest":
                 full_refresh, incremental = RestSalesforceStream, IncrementalRestSalesforceStream
             elif api_type == "bulk":

@@ -44,18 +44,25 @@ connectionSpecification:
     start_date:
       title: Start Date
       description: >-
-        Enter the date in the YYYY-MM-DD format. Airbyte will replicate the data added on and after this date. If this field is blank, Airbyte will replicate the data for last two years.
+        Enter the date (or date-time) in the YYYY-MM-DD or YYYY-MM-DDTHH:mm:ssZ format. Airbyte will replicate the data updated on and after this date. If this field is blank, Airbyte will replicate the data for last two years.
       type: string
       pattern: >-
         ^([0-9]{4}-[0-9]{2}-[0-9]{2}(T[0-9]{2}:[0-9]{2}:[0-9]{2}Z)?)$
+      pattern_descriptor: "YYYY-MM-DD or YYYY-MM-DDTHH:mm:ssZ"
       examples:
         - "2021-07-25"
         - "2021-07-25T00:00:00Z"
       format: date-time
       order: 5
+    force_use_bulk_api:
+      title: Force to use BULK API
+      type: boolean
+      description: Toggle to use Bulk API (this might cause empty fields for some streams)
+      default: false
+      order: 6
     streams_criteria:
       type: array
-      order: 6
+      order: 7
       items:
         type: object
         required:
@@ -81,8 +88,7 @@ connectionSpecification:
             title: Search value
             order: 2
       title: Filter Salesforce Objects
-      description: >-
-        Filter streams relevant to you
+      description: Add filters to select only required stream based on `SObject` name. Use this field to filter which tables are displayed by this connector. This is useful if your Salesforce account has a large number of tables (>1000), in which case you may find it easier to navigate the UI and speed up the connector's performance if you restrict the tables displayed by this connector.
 advanced_auth:
   auth_flow_type: oauth2.0
   predicate_key: