tweaked updated data query to return data types that match the platform message, modify updated data to include platform message meta data, created internal endpoint to run full flow

Author: sposerina

src/server/Dockerfile

@@ -9,7 +9,7 @@ ENV PYTHONUNBUFFERED 1
 
 RUN apt-get update
 
-RUN apt-get install -y python3-dev uwsgi uwsgi-src libcap-dev uwsgi-plugin-python3
+RUN apt-get install -y python3-dev uwsgi uwsgi-src libcap-dev uwsgi-plugin-python3 libpcre3-dev
 RUN pip install --upgrade pip
 
 COPY requirements.txt /

src/server/api/API_ingest/salesforce_contacts.py

@@ -11,6 +11,10 @@
 
 TEST_MODE = os.getenv("TEST_MODE")  # if not present, has value None
 
+DOMAIN = os.getenv("SALESFORCE_DOMAIN")
+CONSUMER_KEY = os.getenv('SALESFORCE_CONSUMER_KEY')
+USERNAME = os.getenv('SALESFORCE_USERNAME')
+
 def store_contacts_all():
     Session = sessionmaker(engine)
     with Session() as session:
@@ -28,8 +32,8 @@ def store_contacts_all():
             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)
+        sf = Salesforce(username=USERNAME, consumer_key=CONSUMER_KEY,
+                        privatekey_file=pem_file, domain=DOMAIN)
         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:
@@ -61,4 +65,4 @@ def store_contacts_all():
                 results = sf.query_more(results['nextRecordsUrl'], True)
         logger.debug("Committing downloaded contact records")
         session.commit()
-    logger.debug("finished downloading latest salesforce contacts data")
+    logger.info("finished downloading latest salesforce contacts data")

src/server/api/API_ingest/updated_data.py

@@ -31,24 +31,24 @@ def get_updated_contact_data():
 
     select json_agg (upd)  as "cd" from (
             select
-            slsf.source_id as "contactId" ,  -- long salesforce string
-            slp.id   as  "personId" ,           -- short PAWS-local shelterluv id
+            slsf.source_id as "Contact_Record_Id__c" ,  -- long salesforce string
+            slp.id   as  "Person_Id__c" ,           -- short PAWS-local shelterluv id
         
-            case
-                when 
-                    (extract(epoch from now())::bigint - foster_out < 365*86400)  -- foster out in last year 
-                    or (extract(epoch from now())::bigint - foster_return < 365*86400) -- foster return 
-                then 'Active'
-                else 'Inactive'
-            end  as "updatedFosterStatus" ,
+            --case
+            --    when 
+            --        (extract(epoch from now())::bigint - foster_out < 365*86400)  -- foster out in last year 
+            --        or (extract(epoch from now())::bigint - foster_return < 365*86400) -- foster return 
+            --    then 'Active'
+            --    else 'Inactive'
+            --end  as "Updated_Recent_Foster_Activity__c",
         
-            (to_timestamp(foster_out ) at time zone  'America/New_York')::date   as "updatedFosterStartDate",
-            (to_timestamp(foster_return ) at time zone 'America/New_York')::date  as "updatedFosterEndDate",
-
-            min(vs.from_date) as "updatedFirstVolunteerDate",
-            max(vs.from_date) as "updatedLastVolunteerDate",
-            vc.source_id as   "volgisticsId"     
+            foster_out as "Updated_Foster_Start_Date__c",
+            foster_return as "Updated_Foster_End_Date__c",
 
+            extract(epoch from min(vs.from_date)) as "Updated_First_Volunteer_Date__c",
+            extract(epoch from max(vs.from_date)) as "Updated_Last_Volunteer_Date__c",
+            sum(vs.hours) as "Updated_Total_Volunteer_Hours__c",
+            vc.source_id::integer as   "Volgistics_Id__c" 
             
         from 
             ev_dates
@@ -69,12 +69,11 @@ def get_updated_contact_data():
             foster_return
         
     ) upd ;
-
-
     """
 
     with Session() as session:
         result = session.execute(qry)
         sfdata = result.fetchone()[0]
+        logger.debug(sfdata)
         logger.debug("Query for Salesforce update returned %d records", len(sfdata))
         return sfdata

src/server/api/internal_api.py

@@ -8,7 +8,7 @@
 from rfm_funcs.create_scores import create_scores
 from api.API_ingest import updated_data
 from pub_sub import salesforce_message_publisher
-import json
+from pipeline import flow_script
 
 logger = structlog.get_logger()
 
@@ -50,17 +50,28 @@ def hit_create_scores():
 @internal_api.route("/api/internal/get_updated_data", methods=["GET"])
 def get_contact_data():
     logger.debug("Calling  get_updated_contact_data()")
-    contact_json =  updated_data.get_updated_contact_data()
-    logger.debug("Returning %d contact records",  len(contact_json) )
+    contact_json = updated_data.get_updated_contact_data()
+    logger.debug("Returning %d contact records", len(contact_json))
     return jsonify(contact_json), 200
 
 
-@internal_api.route("/api/internal/salesforce_platform_message", methods=["POST"])
-def hit_salesforce_platform_message():
-    try:
-        post_dict = json.loads(request.data)
-        salesforce_message_publisher.pipeline_update_message(post_dict)
-    except Exception as e:
-        logger.error(e)
+@internal_api.route("/api/internal/send_salesforce_platform_message", methods=["GET"])
+def send_salesforce_platform_message():
+    contact_list = updated_data.get_updated_contact_data()
+    logger.debug("Returning %d contact records", len(contact_list))
+    salesforce_message_publisher.send_pipeline_update_messages(contact_list)
 
     return jsonify({'outcome': 'OK'}), 200
+
+@internal_api.route("/api/internal/full_flow", methods=["GET"])
+def start_flow():
+    logger.info("Downloading data from APIs")
+    ingest_sources_from_api.start()
+    logger.info("Starting pipeline matching")
+    flow_script.start_flow()
+    logger.info("Building updated data payload")
+    updated_contacts_list = updated_data.get_updated_contact_data()
+    logger.info("Sending Salesforce platform messages")
+    salesforce_message_publisher.send_pipeline_update_messages(updated_contacts_list)
+
+    return jsonify({'outcome': 'OK'}), 200

src/server/pub_sub/salesforce_message_publisher.py

@@ -9,21 +9,21 @@
 import avro.io
 import io
 import structlog
+from datetime import datetime
 
 logger = structlog.get_logger()
 
-# todo: also read these from dict module
 ISSUER = os.getenv("SALESFORCE_CONSUMER_KEY")
 DOMAIN = os.getenv("SALESFORCE_DOMAIN")
 SUBJECT = os.getenv("SALESFORCE_USERNAME")
 INSTANCE_URL = os.getenv("INSTANCE_URL")
 TENANT_ID = os.getenv("TENANT_ID")
+PLATFORM_MESSAGE_AUTHOR = os.getenv("PLATFORM_MESSAGE_AUTHOR_RECORD_ID")
 
 UPDATE_TOPIC = "/event/Updated_Contacts_From_Pipeline__e"
 
 
-def pipeline_update_message(message_dict):
-    # todo: look for certificate using both local and container pathing
+def send_pipeline_update_messages(contacts_list):
     pem_file = 'bin/connected-app-secrets.pem'
     with open(pem_file) as fd:
         private_key = fd.read()
@@ -42,7 +42,6 @@ def pipeline_update_message(message_dict):
         'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
         'assertion': assertion,
     })
-    logger.info(r.json())
     access_token = r.json()['access_token']
     logger.info('Made OAuth call to get access token')
 
@@ -52,16 +51,25 @@ def pipeline_update_message(message_dict):
         auth_meta_data = (('accesstoken', access_token),
                           ('instanceurl', INSTANCE_URL),
                           ('tenantid', TENANT_ID))
+
+
         stub = pb2_grpc.PubSubStub(channel)
         schema_id = stub.GetTopic(pb2.TopicRequest(topic_name=UPDATE_TOPIC), metadata=auth_meta_data).schema_id
         schema = stub.GetSchema(pb2.SchemaRequest(schema_id=schema_id), metadata=auth_meta_data).schema_json
-        buf = io.BytesIO()
-        encoder = avro.io.BinaryEncoder(buf)
-        writer = avro.io.DatumWriter(avro.schema.parse(schema))
-        writer.write(message_dict, encoder)
-        payload = {
-            "schema_id": schema_id,
-            "payload": buf.getvalue()
-        }
-        stub.Publish(pb2.PublishRequest(topic_name=UPDATE_TOPIC, events=[payload]), metadata=auth_meta_data)
-        logger.info('Pipeline update message sent')
+
+        for contact_dict in contacts_list:
+            contact_dict['CreatedDate'] = int(datetime.now().timestamp())
+            contact_dict['CreatedById'] = PLATFORM_MESSAGE_AUTHOR
+
+            buf = io.BytesIO()
+            encoder = avro.io.BinaryEncoder(buf)
+            writer = avro.io.DatumWriter(avro.schema.parse(schema))
+            writer.write(contact_dict, encoder)
+            payload = {
+                "schema_id": schema_id,
+                "payload": buf.getvalue()
+            }
+            stub.Publish(pb2.PublishRequest(topic_name=UPDATE_TOPIC, events=[payload]), metadata=auth_meta_data)
+            logger.info('Pipeline update message sent')
+
+    logger.info("%s total pipeline update messages sent", len(contacts_list))