format by mypy

airbyte-integrations/connectors/source-salesforce/integration_tests/integration_test.py

@@ -4,6 +4,7 @@
 
 import base64
 import json
+import time
 from datetime import datetime
 from pathlib import Path
 
@@ -71,6 +72,8 @@ def get_stream_state():
 
 def test_update_for_deleted_record(stream):
     headers = stream.authenticator.get_auth_header()
+    stream_state = get_stream_state()
+    time.sleep(1)
     response = create_note(stream, headers)
     assert response.status_code == 201, "Note was note created"
 
@@ -79,7 +82,6 @@ def test_update_for_deleted_record(stream):
     notes = set(record["Id"] for record in stream.read_records(sync_mode=None))
     assert created_note_id in notes, "No created note during the sync"
 
-    stream_state = get_stream_state()
     response = delete_note(stream, created_note_id, headers)
     assert response.status_code == 204, "Note was not deleted"
 
@@ -94,6 +96,7 @@ def test_update_for_deleted_record(stream):
     assert is_deleted, "Wrong field value for deleted note during the sync"
 
     stream_state = get_stream_state()
+    time.sleep(1)
     response = update_note(stream, created_note_id, headers)
     assert response.status_code == 404, "Note was updated, but should not"
 

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

@@ -5,11 +5,11 @@
 import concurrent.futures
 from typing import Any, List, Mapping, Optional, Tuple
 
-import requests
+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
+from requests.exceptions import HTTPError, RequestException  # type: ignore[import]
 
 from .exceptions import TypeSalesforceException
 from .rate_limiting import default_backoff_handler
@@ -185,14 +185,14 @@ def __init__(
         client_secret: str = None,
         is_sandbox: bool = None,
         start_date: str = None,
-        **kwargs,
-    ):
+        **kwargs: Any,
+    ) -> None:
         self.refresh_token = refresh_token
         self.token = token
         self.client_id = client_id
         self.client_secret = client_secret
         self.access_token = None
-        self.instance_url = None
+        self.instance_url = ""
         self.session = requests.Session()
         # Change the connection pool size. Default value is not enough for parallel tasks
         adapter = request_adapters.HTTPAdapter(pool_connections=self.parallel_tasks_size, pool_maxsize=self.parallel_tasks_size)
@@ -203,7 +203,7 @@ def __init__(
             self.logger.info("using SANDBOX of Salesforce")
         self.start_date = start_date
 
-    def _get_standard_headers(self):
+    def _get_standard_headers(self) -> Mapping[str, str]:
         return {"Authorization": "Bearer {}".format(self.access_token)}
 
     def get_streams_black_list(self) -> List[str]:
@@ -287,34 +287,32 @@ def describe(self, sobject: str = None, sobject_options: Mapping[str, Any] = Non
         resp = self._make_request("GET", url, headers=headers)
         if resp.status_code == 404 and sobject:
             self.logger.error(f"not found a description for the sobject '{sobject}'. Sobject options: {sobject_options}")
-        return resp.json()
+        resp_json: Mapping[str, Any] = resp.json()
+        return resp_json
 
     def generate_schema(self, stream_name: str = None, stream_options: Mapping[str, Any] = None) -> Mapping[str, Any]:
         response = self.describe(stream_name, stream_options)
         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)
+            schema["properties"][field["name"]] = self.field_to_property_schema(field)  # type: ignore[index]
         return schema
 
     def generate_schemas(self, stream_objects: Mapping[str, Any]) -> Mapping[str, Any]:
-        def load_schema(name: str, stream_options: Mapping[str, Any]) -> Tuple[Mapping[str, Any], Optional[str]]:
+        def load_schema(name: str, stream_options: Mapping[str, Any]) -> Tuple[str, Optional[Mapping[str, Any]], Optional[str]]:
             try:
                 result = self.generate_schema(stream_name=name, stream_options=stream_options)
             except RequestException as e:
-                return None, str(e)
-            return result, None
+                return name, None, str(e)
+            return name, result, None
 
         stream_names = list(stream_objects.keys())
         # try to split all requests by chunks
         stream_schemas = {}
         for i in range(0, len(stream_names), self.parallel_tasks_size):
             chunk_stream_names = stream_names[i : i + self.parallel_tasks_size]
             with concurrent.futures.ThreadPoolExecutor(max_workers=len(chunk_stream_names)) as executor:
-                for stream_name, (schema, err) in zip(
-                    chunk_stream_names,
-                    executor.map(
-                        lambda args: load_schema(*args), [(stream_name, stream_objects[stream_name]) for stream_name in chunk_stream_names]
-                    ),
+                for stream_name, schema, err in executor.map(
+                    lambda args: load_schema(*args), [(stream_name, stream_objects[stream_name]) for stream_name in chunk_stream_names]
                 ):
                     if err:
                         self.logger.error(f"Loading error of the {stream_name} schema: {err}")
@@ -347,13 +345,16 @@ def field_to_property_schema(field_params: Mapping[str, Any]) -> Mapping[str, An
         if sf_type in STRING_TYPES:
             property_schema["type"] = ["string", "null"]
         elif sf_type in DATE_TYPES:
-            property_schema = {"type": ["string", "null"], "format": "date-time" if sf_type == "datetime" else "date"}
+            property_schema = {
+                "type": ["string", "null"],
+                "format": "date-time" if sf_type == "datetime" else "date",  # type: ignore[dict-item]
+            }
         elif sf_type in NUMBER_TYPES:
             property_schema["type"] = ["number", "null"]
         elif sf_type == "address":
             property_schema = {
                 "type": ["object", "null"],
-                "properties": {
+                "properties": {  # type: ignore[dict-item]
                     "street": {"type": ["null", "string"]},
                     "state": {"type": ["null", "string"]},
                     "postalCode": {"type": ["null", "string"]},
@@ -365,7 +366,7 @@ def field_to_property_schema(field_params: Mapping[str, Any]) -> Mapping[str, An
                 },
             }
         elif sf_type == "base64":
-            property_schema = {"type": ["string", "null"], "format": "base64"}
+            property_schema = {"type": ["string", "null"], "format": "base64"}  # type: ignore[dict-item]
         elif sf_type == "int":
             property_schema["type"] = ["integer", "null"]
         elif sf_type == "boolean":
@@ -379,7 +380,10 @@ def field_to_property_schema(field_params: Mapping[str, Any]) -> Mapping[str, An
         elif sf_type == "location":
             property_schema = {
                 "type": ["object", "null"],
-                "properties": {"longitude": {"type": ["null", "number"]}, "latitude": {"type": ["null", "number"]}},
+                "properties": {  # type: ignore[dict-item]
+                    "longitude": {"type": ["null", "number"]},
+                    "latitude": {"type": ["null", "number"]},
+                },
             }
         else:
             raise TypeSalesforceException("Found unsupported type: {}".format(sf_type))

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

@@ -8,7 +8,7 @@
 import backoff
 from airbyte_cdk.logger import AirbyteLogger
 from airbyte_cdk.sources.streams.http.exceptions import DefaultBackoffException
-from requests import codes, exceptions
+from requests import codes, exceptions  # type: ignore[import]
 
 TRANSIENT_EXCEPTIONS = (
     DefaultBackoffException,

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

@@ -3,15 +3,15 @@
 #
 
 import copy
-from typing import Any, Iterator, List, Mapping, MutableMapping, Tuple
+from typing import Any, Iterator, List, Mapping, MutableMapping, Optional, Tuple
 
 from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.models import AirbyteMessage, ConfiguredAirbyteCatalog
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
 from airbyte_cdk.sources.utils.schema_helpers import split_config
-from requests import codes, exceptions
+from requests import codes, exceptions  # type: ignore[import]
 
 from .api import UNSUPPORTED_BULK_API_SALESFORCE_OBJECTS, UNSUPPORTED_FILTERING_STREAMS, Salesforce
 from .streams import BulkIncrementalSalesforceStream, BulkSalesforceStream, IncrementalSalesforceStream, SalesforceStream
@@ -24,17 +24,18 @@ def _get_sf_object(config: Mapping[str, Any]) -> Salesforce:
         sf.login()
         return sf
 
-    def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, any]:
+    def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, Optional[str]]:
         try:
             _ = self._get_sf_object(config)
-            return True, None
         except exceptions.HTTPError as error:
             error_data = error.response.json()[0]
             error_code = error_data.get("errorCode")
             if error.response.status_code == codes.FORBIDDEN and error_code == "REQUEST_LIMIT_EXCEEDED":
                 logger.warn(f"API Call limit is exceeded. Error message: '{error_data.get('message')}'")
                 return False, "API Call limit is exceeded"
 
+        return True, None
+
     @classmethod
     def generate_streams(
         cls,
@@ -80,7 +81,11 @@ def streams(self, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog =
         return self.generate_streams(config, stream_objects, sf, state=state)
 
     def read(
-        self, logger: AirbyteLogger, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog, state: MutableMapping[str, Any] = None
+        self,
+        logger: AirbyteLogger,
+        config: Mapping[str, Any],
+        catalog: ConfiguredAirbyteCatalog,
+        state: Optional[MutableMapping[str, Any]] = None,
     ) -> Iterator[AirbyteMessage]:
         """
         Overwritten to dynamically receive only those streams that are necessary for reading for significant speed gains