fix(ingest/salesforce): support JSON web token auth (datahub-project#…

…7963)

metadata-ingestion/docs/sources/salesforce/salesforce_pre.md

@@ -1,8 +1,9 @@
 ### Prerequisites
 
-In order to ingest metadata from Salesforce, you will need:
+In order to ingest metadata from Salesforce, you will need one of:
 
-- Salesforce username, password, [security token](https://developer.Salesforce.com/docs/atlas.en-us.api.meta/api/sforce_api_concepts_security.htm) OR 
+- Salesforce username, password, [security token](https://developer.Salesforce.com/docs/atlas.en-us.api.meta/api/sforce_api_concepts_security.htm)
+- Salesforce username, consumer key and private key for [JSON web token access](https://help.salesforce.com/s/articleView?id=sf.remoteaccess_oauth_jwt_flow.htm&type=5)
 - Salesforce instance url and access token/session id (suitable for one-shot ingestion only, as access token typically expires after 2 hours of inactivity)
 
 The account used to access Salesforce requires the following permissions for this integration to work:

metadata-ingestion/src/datahub/ingestion/source/salesforce.py

@@ -61,6 +61,7 @@
 class SalesforceAuthType(Enum):
     USERNAME_PASSWORD = "USERNAME_PASSWORD"
     DIRECT_ACCESS_TOKEN = "DIRECT_ACCESS_TOKEN"
+    JSON_WEB_TOKEN = "JSON_WEB_TOKEN"
 
 
 class SalesforceProfilingConfig(ConfigModel):
@@ -80,9 +81,9 @@ class SalesforceConfig(DatasetSourceConfigMixin):
     # Username, Password Auth
     username: Optional[str] = Field(description="Salesforce username")
     password: Optional[str] = Field(description="Password for Salesforce user")
-    security_token: Optional[str] = Field(
-        description="Security token for Salesforce username"
-    )
+    consumer_key: Optional[str] = Field(description="Consumer key for Salesforce JSON web token access")
+    private_key: Optional[str] = Field(description="Private key as a string for Salesforce JSON web token access")
+    security_token: Optional[str] = Field(description="Security token for Salesforce username")
     # client_id, client_secret not required
 
     # Direct - Instance URL, Access Token Auth
@@ -230,6 +231,26 @@ def __init__(self, config: SalesforceConfig, ctx: PipelineContext) -> None:
                     domain="test" if self.config.is_sandbox else None,
                 )
 
+            elif self.config.auth is SalesforceAuthType.JSON_WEB_TOKEN:
+                logger.debug("Json Web Token provided in the config")
+                assert (
+                    self.config.username is not None
+                ), "Config username is required for JSON_WEB_TOKEN auth"
+                assert (
+                    self.config.consumer_key is not None
+                ), "Config consumer_key is required for JSON_WEB_TOKEN auth"
+                assert (
+                    self.config.private_key is not None
+                ), "Config private_key is required for JSON_WEB_TOKEN auth"
+
+                self.sf = Salesforce(
+                    username=self.config.username,
+                    consumer_key=self.config.consumer_key,
+                    privatekey=self.config.private_key,
+                    session=self.session,
+                    domain="test" if self.config.is_sandbox else None,
+                )
+
         except Exception as e:
             logger.error(e)
             raise ConfigurationError("Salesforce login failed") from e