Merge pull request #560 from CodeForPhilly/506-salesforce-message-publisher

506 salesforce message publisher

Author: c-simpson

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

@@ -1,9 +1,9 @@
 
-from sqlalchemy.orm import  sessionmaker
-from simple_salesforce import Salesforce
+import structlog
+from sqlalchemy.orm import sessionmaker
+
 from config import engine
 
-import structlog
 logger = structlog.get_logger()
 
 
@@ -14,10 +14,10 @@ def get_updated_contact_data():
         with ev_dates as 
         (select 
             person_id, 
-            max(case when event_type=1 then time else null end) adopt,
-            max(case when event_type=2 then time else null end) foster_out,
+            max(case when event_type=1 then time else null end) * 1000 adopt,
+            max(case when event_type=2 then time else null end) * 1000 foster_out,
            --  max(case when event_type=3 then time else null end) rto,
-            max(case when event_type=5 then time else null end) foster_return
+            max(case when event_type=5 then time else null end) * 1000 foster_return
            
         from 
             sl_animal_events sla
@@ -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.internal_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)) * 1000 as "Updated_First_Volunteer_Date__c",
+            extract(epoch from max(vs.from_date)) * 1000 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
@@ -63,18 +63,17 @@ def get_updated_contact_data():
 
         group by
             slsf.source_id,
-            slp.id,
+            slp.internal_id,
             vc.source_id,
             foster_out ,
             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

@@ -3,10 +3,14 @@
 import structlog
 from flask import jsonify
 
-from api.API_ingest import ingest_sources_from_api, salesforce_contacts
-from api.api import internal_api
-# from rfm_funcs.create_scores import create_scores
+from api.API_ingest import ingest_sources_from_api
 from api.API_ingest import updated_data
+from api.api import internal_api
+
+from pipeline import flow_script
+from pub_sub import salesforce_message_publisher
+from rfm_funcs.create_scores import create_scores
+
 
 logger = structlog.get_logger()
 
@@ -48,6 +52,28 @@ def ingest_raw_data():
 @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/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/__init__.py

@@ -0,0 +1,75 @@
+import time
+import jwt
+import os
+import requests
+import certifi
+import grpc
+import pub_sub.stubs.pubsub_api_pb2_grpc as pb2_grpc
+import pub_sub.stubs.pubsub_api_pb2 as pb2
+import avro.io
+import io
+import structlog
+from datetime import datetime
+
+logger = structlog.get_logger()
+
+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 send_pipeline_update_messages(contacts_list):
+    pem_file = 'bin/connected-app-secrets.pem'
+    with open(pem_file) as fd:
+        private_key = fd.read()
+    logger.info('Loaded PEM certificate')
+
+    claim = {
+        'iss': ISSUER,
+        'exp': int(time.time()) + 300,
+        'aud': 'https://{}.salesforce.com'.format(DOMAIN),
+        'sub': SUBJECT,
+    }
+    assertion = jwt.encode(claim, private_key, algorithm='RS256', headers={'alg': 'RS256'})
+    logger.info('Generated JWT')
+
+    r = requests.post('https://{}.salesforce.com/services/oauth2/token'.format(DOMAIN), data={
+        'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
+        'assertion': assertion,
+    })
+    access_token = r.json()['access_token']
+    logger.info('Made OAuth call to get access token')
+
+    with open(certifi.where(), 'rb') as f:
+        creds = grpc.ssl_channel_credentials(f.read())
+    with grpc.secure_channel('api.pubsub.salesforce.com:7443', creds) as channel:
+        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
+
+        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))