remove dependency on dropbox for getting Shelterluv data, make sure data pull is all done in the same transaction

Author: sposerina

src/server/api/API_ingest/ingest_sources_from_api.py

@@ -1,9 +1,10 @@
-from api.API_ingest import shelterluv_api_handler
+from api.API_ingest import shelterluv_api_handler, salesforce_api_handler
 import structlog
 logger = structlog.get_logger()
 
-def start(conn):
+def start(session):
     logger.debug("Start Fetching raw data from different API sources")
     #Run each source to store the output in dropbox and in the container as a CSV
-    shelterluv_api_handler.store_shelterluv_people_all(conn)
+    shelterluv_api_handler.store_shelterluv_people_all(session)
+    salesforce_api_handler.store_contacts_all(session)
     logger.debug("Finish Fetching raw data from different API sources")

src/server/api/API_ingest/salesforce_api_handler.py

@@ -5,11 +5,8 @@
 from config import engine
 from models import SalesForceContacts
 
-def ingest_data():
+def store_contacts_all(session):
 
-    Session = sessionmaker(engine)
-
-    session = Session()
     session.execute("TRUNCATE TABLE salesforcecontacts")
 
     sf = Salesforce(domain=os.getenv('SALESFORCE_DOMAIN'), password=os.getenv('SALESFORCE_PW'), username=os.getenv('SALESFORCE_USERNAME'), organizationId=os.getenv('SALESFORCE_ORGANIZATION_ID'), security_token=os.getenv('SALESFORCE_SECURITY_TOKEN'))
@@ -33,6 +30,4 @@ def ingest_data():
             session.add(contact)
             done = results['done']
             if not done:
-                results = sf.query_more(results['nextRecordsUrl'])
-
-    session.commit()
+                results = sf.query_more(results['nextRecordsUrl'])

src/server/api/API_ingest/shelterluv_api_handler.py

@@ -1,11 +1,4 @@
-import csv
-import os
-import time
-
-import requests
-import pandas as pd
-from api.API_ingest.dropbox_handler import upload_file_to_dropbox
-from constants import RAW_DATA_PATH
+import requests, os
 from models import ShelterluvPeople
 import structlog
 logger = structlog.get_logger()
@@ -25,31 +18,8 @@
         logger.error("Couldn't get SHELTERLUV_SECRET_TOKEN from file or environment")
 
 
-def write_csv(json_data):
-    now = time.localtime()
-    now_date = time.strftime("%Y-%m-%d--%H-%M-%S", now)
-
-    path = RAW_DATA_PATH + "shelterluvpeople-" + now_date + ".csv"  # store file name to use for dropbox
-
-    file_handle = open(path, "w")
-
-    csv_writer = csv.writer(file_handle)
-
-    count = 0
-    for item in json_data:
-        if count == 0:
-            # Writing headers of CSV file
-            header = item.keys()
-            csv_writer.writerow(header)
-            count += 1
-
-        # Writing data of CSV file
-        csv_writer.writerow(item.values())
-
-    file_handle.close()
-
-    return path
 
+TEST_MODE=os.getenv("TEST_MODE")  # if not present, has value None
 #################################
 # This script is used to fetch data from shelterluv API.
 # Please be mindful of your usage.
@@ -64,39 +34,37 @@ def write_csv(json_data):
 
 ''' Iterate over all shelterlove people and store in json file in the raw data folder
 We fetch 100 items in each request, since that is the limit based on our research '''
-def store_shelterluv_people_all(conn):
+def store_shelterluv_people_all(session):
     offset = 0
     LIMIT = 100
     has_more = True
-    shelterluv_people = []
+
+    session.execute("TRUNCATE TABLE shelterluvpeople")
 
     logger.debug("Start getting shelterluv contacts from people table")
 
     while has_more:
         r = requests.get("http://shelterluv.com/api/v1/people?limit={}&offset={}".format(LIMIT, offset),
                          headers={"x-api-key": SHELTERLUV_SECRET_TOKEN})
         response = r.json()
-        shelterluv_people += response["people"]
-        has_more = response["has_more"]
-        offset += 100
-
-    logger.debug("Finish getting shelterluv contacts from people table")
+        for person in response["people"]:
+            #todo: Does this need more "null checks"?
+            session.add(ShelterluvPeople(firstname=person["Firstname"],
+                                  lastname=person["Lastname"],
+                                  id=person["ID"] if "ID" in person else None,
+                                  internal_id=person["Internal-ID"],
+                                  associated=person["Associated"],
+                                  street=person["Street"],
+                                  apartment=person["Apartment"],
+                                  city=person["City"],
+                                  state=person["State"],
+                                  zip=person["Zip"],
+                                  email=person["Email"],
+                                  phone=person["Phone"],
+                                  animal_ids=person["Animal_ids"]))
+        offset += LIMIT
+        has_more = response["has_more"] if not TEST_MODE else response["has_more"] and offset < 1000
 
-    logger.debug("Start storing latest shelterluvpeople results to container")
-    if os.listdir(RAW_DATA_PATH):
-        for file_name in os.listdir(RAW_DATA_PATH):
-            file_path = os.path.join(RAW_DATA_PATH, file_name)
-            file_name_striped = file_path.split('-')[0].split('/')[-1]
 
-            if file_name_striped == "shelterluvpeople":
-                os.remove(file_path)
-
-    file_path = write_csv(shelterluv_people)
-    logger.debug("Finish storing latest shelterluvpeople results to container")
-
-    logger.debug("Start storing " + '/shelterluv/' + "results to dropbox")
-    upload_file_to_dropbox(file_path, '/shelterluv/' + file_path.split('/')[-1])
-    logger.debug("Finish storing " + '/shelterluv/' + "results to dropbox")
+    logger.debug("Finish getting shelterluv contacts from people table")
 
-    logger.debug("Uploading shelterluvpeople csv to database")
-    ShelterluvPeople.insert_from_df(pd.read_csv(file_path, dtype="string"), conn)

src/server/api/admin_api.py

@@ -9,13 +9,12 @@
 from sqlalchemy import Table, MetaData
 from pipeline import flow_script
 from config import engine
-from flask import request, redirect, jsonify, current_app
+from flask import request, redirect, jsonify
 from api.file_uploader import validate_and_arrange_upload
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import sessionmaker
 
 from api import jwt_ops
 from config import RAW_DATA_PATH
-from api.API_ingest.salesforce_api_handler import ingest_data
 
 import structlog
 logger = structlog.get_logger()

src/server/api/internal_api.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from api.API_ingest import ingest_sources_from_api
 from rfm_funcs.create_scores import create_scores
+from sqlalchemy.orm import  sessionmaker
 
 import structlog
 logger = structlog.get_logger()
@@ -28,8 +29,10 @@ def user_test2():
 @internal_api.route("/api/ingestRawData", methods=["GET"])
 def ingest_raw_data():
     try:
-        with engine.begin() as conn:
-            ingest_sources_from_api.start(conn)
+        Session = sessionmaker(engine)
+        with Session() as session:
+            ingest_sources_from_api.start(session)
+            session.commit()
     except Exception as e:
         logger.error(e)
 

src/server/secrets_dict.py

@@ -1,5 +1,4 @@
 SD_COMMENT="This is for local development"
-SHELTERLUV_SECRET_TOKEN=""
 APP_SECRET_KEY="ASKASK"
 JWT_SECRET="JWTSECRET"
 POSTGRES_PASSWORD="thispasswordisverysecure"