Merge branch 'table_for_file' of https://github.com/CodeForPhilly/paws-data-pipeline into execute-adjustments

� Conflicts:
�	src/server/pipeline/match_data.py

Author: urirot

src/server/api/admin_api.py

@@ -131,6 +131,20 @@ def list_statistics():
     return last_execution_details
 
 
+@admin_api.route("/api/get_execution_status/<int:job_id>", methods=["GET"])
+def get_exec_status(job_id):
+    kvt = Table("kv_unique", metadata, autoload=True, autoload_with=engine)
+    with engine.connect() as connection:
+        s_jobid = 'job-' + str(job_id)        
+        s = text("select valcol from kv_unique where keycol = :j ;")
+        s = s.bindparams(j=s_jobid)
+        result = connection.execute(s)
+        exec_status  = result.fetchone()[0]
+
+    return exec_status
+
+
+
 """
 @admin_api.route('/api/status', methods=['GET'])
 def checkStatus():

src/server/app.py

@@ -16,6 +16,7 @@
 app.secret_key = APP_SECRET_KEY
 app.config["MAX_CONTENT_LENGTH"] = 500 * 1024 * 1024  # 500 Megs
 app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 0
+
 from api.admin_api import admin_api
 from api.common_api import common_api
 from api.user_api import user_api
@@ -24,6 +25,8 @@
 app.register_blueprint(common_api)
 app.register_blueprint(user_api)
 
+app.logger.setLevel('INFO')
+
 # init_db_schema.start(connection)
 
 

src/server/config.py

@@ -76,10 +76,3 @@
 os.makedirs(CURRENT_SOURCE_FILES_PATH, exist_ok=True)
 os.makedirs(REPORT_PATH, exist_ok=True)
 os.makedirs(ZIPPED_FILES, exist_ok=True)
-
-if not (os.path.exists(LOGS_PATH + "last_execution.json")):
-    f = open(
-        LOGS_PATH + "last_execution.json", "w"
-    )  # Prevent 500 error from /api/statistics
-    f.write("{}")
-    f.close()

src/server/pipeline/log_db.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+import json
+from sqlalchemy.sql import text
+from flask import  current_app
+
+from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey, exc, select
+
+from config import engine
+
+
+metadata = MetaData()
+
+kvt = Table("kv_unique", metadata, autoload=True, autoload_with=engine)
+
+
+
+def log_exec_status(job_id: str, job_status: dict):
+
+    # Write Last Execution stats to DB  
+    # See Alembic Revision ID: 05e0693f8cbb for table definition
+    with engine.connect() as connection:
+        ins_stmt = insert(kvt).values(               # Postgres-specific insert() supporting ON CONFLICT 
+            keycol = 'job-' + job_id,
+            valcol = json.dumps(job_status)
+            )
+
+        # If key already present in DB, do update instead 
+        upsert = ins_stmt.on_conflict_do_update(
+                constraint='kv_unique_keycol_key',
+                set_=dict(valcol=json.dumps(job_status))
+                )
+
+        try:
+            connection.execute(upsert)
+        except Exception as e:
+            current_app.logger.error("Insert/Update failed Execution status")
+            current_app.logger.exception(e)
+
+

src/server/pipeline/match_data.py

@@ -1,8 +1,10 @@
-import datetime
+import datetime, time
 import pandas as pd
 import numpy as np
 
 from flask import current_app
+from pipeline import log_db
+
 
 
 def start(connection, added_or_updated_rows):
@@ -12,6 +14,10 @@ def start(connection, added_or_updated_rows):
     current_app.logger.info('Start record matching')
     # Will need to consider updating the existing row contents (filter by active), deactivate,
     # try to match, and merge previous matching groups if applicable
+
+    job_id = str(int(time.time()))
+    log_db.log_exec_status(job_id,{'status': 'starting', 'at_row':  0, 'of_rows':0})
+    current_app.logger.info("Running execute job ID " + job_id)
     items_to_update = pd.concat([added_or_updated_rows["new"], added_or_updated_rows["updated"]], ignore_index=True)
     pdp_contacts = pd.read_sql_table('pdp_contacts', connection)
 
@@ -30,8 +36,9 @@ def start(connection, added_or_updated_rows):
     for row_num, row in enumerate(rows):
         if row_num % row_print_freq == 0:
             current_app.logger.info("- Matching rows {}-{} of {}".format(
-                row_num + 1, min(len(rows), row_num + row_print_freq), len(rows))
+                row_num+1, min(len(rows), row_num+row_print_freq), len(rows))
             )
+            log_db.log_exec_status(job_id,{'status': 'executing', 'at_row':  row_num+1, 'of_rows':len(rows)})
 
         # Exact matches based on specified columns
         row_matches = pdp_contacts[
@@ -66,3 +73,5 @@ def start(connection, added_or_updated_rows):
     current_app.logger.info("- Writing data to pdp_contacts table")
     items_to_update.to_sql('pdp_contacts', connection, index=False, if_exists='append')
     current_app.logger.info("- Finished load to pdp_contacts table")
+
+    log_db.log_exec_status(job_id,{'status': 'complete', 'at_row':  len(rows), 'of_rows':len(rows)})