Merge remote-tracking branch 'origin/master' into 510-sf-updated

Author: sposerina

.gitignore

@@ -23,3 +23,5 @@ start_env.sh
 *secrets*
 *kustomization*
 src/.venv/
+src/server/secrets_dict.py
+

src/server/api/.optic/.gitignore

@@ -10,8 +10,8 @@ def start():
     logger.debug("   Finished fetching Salesforce contacts")
 
     logger.debug("   Fetching Shelterluv people")
-    shelterluv_people.store_shelterluv_people_all()
-    logger.debug("   Finished fetching Shelterluv people")
+    slp_count = shelterluv_people.store_shelterluv_people_all()
+    logger.debug("   Finished fetching Shelterluv people - %d records" , slp_count)
 
     logger.debug("   Fetching Shelterluv events")
     sle_count = sl_animal_events.store_all_animals_and_events()

src/server/api/API_ingest/ingest_sources_from_api.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,10 +19,27 @@ 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 os.path.exists('server/bin/connected-app-secrets.pem'):
+            pem_file = 'server/bin/connected-app-secrets.pem'
+        elif os.path.exists('bin/connected-app-secrets.pem'):
+            pem_file = 'bin/connected-app-secrets.pem'
+        else:
+            logger.error("Missing salesforce jwt private key pem file, skipping data pull")
+            return
+
+        sf = Salesforce(username=os.getenv('SALESFORCE_USERNAME'), consumer_key=os.getenv('SALESFORCE_CONSUMER_KEY'),
+                        privatekey_file=pem_file)
         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("%d total Salesforce contact records", results['totalSize'])
+        if TEST_MODE:
+            logger.debug("running in test mode so only downloading first page of Salesforce contacts")
+
+        total_records = 0
         done = False
         while not done:
+            total_records += len(results['records'])
+            logger.debug("Query returned %d Salesforce contact records, total %d", len(results['records']), total_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'],
@@ -33,10 +53,12 @@ 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)
+        logger.debug("Committing downloaded contact records")
         session.commit()
-    logger.debug("finished downloading latest salesforce contacts data")
+    logger.debug("finished downloading latest salesforce contacts data")

src/server/api/API_ingest/salesforce_contacts.py

@@ -70,7 +70,10 @@ def store_shelterluv_people_all():
                                       animal_ids=person["Animal_ids"]))
             offset += LIMIT
             has_more = response["has_more"] if not TEST_MODE else response["has_more"] and offset < 1000
+            if offset % 1000 == 0:
+                logger.debug("Reading offset %s", str(offset))
         session.commit()
 
-    logger.debug("Finish getting shelterluv contacts from people table")
+    logger.debug("Finished getting shelterluv contacts from people table")
+    return offset
 

src/server/api/API_ingest/shelterluv_people.py

@@ -224,10 +224,10 @@ def start_job():
 
     if running_job :
         # There was a running job already
-        logger.warn("Request to start job, but job_id " + str(running_job) + " already executing")
+        logger.warn("Request to start job, but job_id %s already executing", str(running_job))
         return None
     else:
-        logger.info("Assigned job_id  %s" + str(job_id ) )
+        logger.info("Assigned job_id  %s" ,  str(job_id ) )
         return job_id
 
 

src/server/api/admin_api.py

@@ -67,22 +67,26 @@ def start_flow():
             # 6. Update each row in pdp_contacts to give it a match id
             #    corresponding to its connected componenet.
 
-            logger.info("Clearing pdp_contacts to prepare for match")
+            logger.debug("Clearing pdp_contacts to prepare for match")
             reset_pdp_contacts_with_unmatched(conn)
 
-            logger.info("Computing matches")
+            logger.debug("Computing automatic matches")
             automatic_matches = get_automatic_matches(conn)
+            logger.debug("Computing manual matches")
             manual_matches = get_manual_matches(conn)
 
             match_graph = Graph()
+            logger.debug("Adding automatic matches to graph")
             match_graph.add_edges_from(automatic_matches)
+            logger.debug("Adding manual matches to graph")
             match_graph.add_edges_from(manual_matches)
+            logger.debug("Processing graph")
             match_groups = connected_components(match_graph)
 
-            logger.info("Updating pdp_contacts with match ids")
+            logger.debug("Updating pdp_contacts with match ids")
             update_matching_ids(match_groups, conn)
 
-            logger.info("Finished flow script run")
+            logger.debug("Finished flow script run")
             job_outcome = "completed"
             log_db.log_exec_status(job_id, "flow", "complete", "")
 

src/server/pipeline/flow_script.py

@@ -130,7 +130,7 @@ def validate_import_vs(filename, conn):
         # parent will commit. Don't commit here!
 
 
-        logger.info("Total rows: %s  Dupes: %s Missing volgistics id: ",  str(row_count), str(dupes), str(missing_volgistics_id)  )
+        logger.info("Total rows: %s  Dupes: %s Missing volgistics id: %s",  str(row_count), str(dupes), str(missing_volgistics_id)  )
         logger.info("Other integrity exceptions: %s  Other exceptions: %s",  str(other_exceptions),  str(other_integrity) )
         wb.close()
         return { True : "File imported" }