changed Salesforce authentication to the JWT flow, added test mode to return a subset of the contacts

Author: sposerina

src/server/api/API_ingest/salesforce_contacts.py

@@ -1,13 +1,16 @@
 import os
 
-from sqlalchemy.orm import  sessionmaker
+import structlog
 from simple_salesforce import Salesforce
+from sqlalchemy.orm import sessionmaker
+
 from config import engine
 from models import SalesForceContacts
 
-import structlog
 logger = structlog.get_logger()
 
+TEST_MODE = os.getenv("TEST_MODE")  # if not present, has value None
+
 def store_contacts_all():
     Session = sessionmaker(engine)
     with Session() as session:
@@ -16,12 +19,21 @@ def store_contacts_all():
         session.execute("TRUNCATE TABLE salesforcecontacts")
 
         logger.debug("retrieving the latest salesforce contacts data")
-        sf = Salesforce(domain=os.getenv('SALESFORCE_DOMAIN'), password=os.getenv('SALESFORCE_PW'), username=os.getenv('SALESFORCE_USERNAME'), security_token=os.getenv('SALESFORCE_SECURITY_TOKEN'))
+
+        if not os.path.exists('bin/connected-app-secrets.pem'):
+            logger.warn("missing salesforce jwt private key, skipping data pull")
+            return
+
+        sf = Salesforce(username=os.getenv('SALESFORCE_USERNAME'), consumer_key=os.getenv('SALESFORCE_CONSUMER_KEY'),
+                        privatekey_file='bin/connected-app-secrets.pem')
         results = sf.query("SELECT Contact_ID_18__c, FirstName, LastName, Contact.Account.Name, MailingCountry, MailingStreet, MailingCity, MailingState, MailingPostalCode, Phone, MobilePhone, Email FROM Contact")
-        logger.debug("Query returned %d Salesforce contact records", len(results['records']) )
-        
+        logger.debug("%d total Salesforce contact records", results['totalSize'])
+        if TEST_MODE:
+            logger.debug("running in test mode so only downloading first page of Salesforce contacts")
+
         done = False
         while not done:
+            logger.debug("Query returned %d Salesforce contact records", len(results['records']))
             for row in results['records']:
                 account_name = row['Account']['Name'] if row['Account'] is not None else None
                 contact = SalesForceContacts(contact_id=row['Contact_ID_18__c'],
@@ -35,10 +47,11 @@ def store_contacts_all():
                                              mailing_zip_postal_code=row['MailingPostalCode'],
                                              phone=row['Phone'],
                                              mobile=row['MobilePhone'],
-                                             email=['Email'])
+                                             email=row['Email'])
                 session.add(contact)
-                done = results['done']
-                if not done:
-                    results = sf.query_more(results['nextRecordsUrl'])
+            # if in test mode only return first page of results
+            done = results['done'] if not TEST_MODE else True
+            if not done:
+                results = sf.query_more(results['nextRecordsUrl'], True)
         session.commit()
-    logger.debug("finished downloading latest salesforce contacts data")
+    logger.debug("finished downloading latest salesforce contacts data")