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

� Conflicts:
�	src/server/api/admin_api.py

Author: urirot

src/server/alembic/versions/05e0693f8cbb_key_value_table.py

@@ -0,0 +1,31 @@
+"""key/value table
+
+Revision ID: 05e0693f8cbb
+Revises: 6b8cf99be000
+Create Date: 2021-03-18 11:35:43.512082
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '05e0693f8cbb'
+down_revision = '6b8cf99be000'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'kv_unique',
+        sa.Column('_id', sa.Integer, primary_key=True),
+        sa.Column('keycol', sa.String(50), nullable=False, unique=True),
+        sa.Column('valcol', sa.String(65536), nullable=True),
+    )
+    
+    # op.create_index('kvk_ix', 'kv_unique', ['key'], unique=True)
+
+
+def downgrade():
+    pass

src/server/api/admin_api.py

@@ -1,8 +1,12 @@
 from api.api import admin_api
+import shutil
 import os
 from datetime import datetime
 import json
 from sqlalchemy.sql import text
+
+from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey, exc, select
 from pipeline import flow_script
 from config import engine
 from flask import request, redirect, jsonify, current_app, abort
@@ -15,10 +19,14 @@
 
 ALLOWED_EXTENSIONS = {"csv", "xlsx"}
 
+metadata = MetaData()
 
 def __allowed_file(filename):
     return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS
 
+kvt = Table("kv_unique", metadata, autoload=True, autoload_with=engine)
+
+
 
 # file upload tutorial
 @admin_api.route("/api/file", methods=["POST"])
@@ -36,7 +44,7 @@ def uploadCSV():
             finally:
                 file.close()
 
-    return jsonify({"success": "uploaded file"});
+    return redirect("/")
 
 
 @admin_api.route("/api/listCurrentFiles", methods=["GET"])
@@ -55,38 +63,38 @@ def list_current_files():
 @admin_api.route("/api/execute", methods=["GET"])
 def execute():
     current_app.logger.info("Execute flow")
+    flow_script.start_flow()
 
-    try:
-        last_execution_file = open(LOGS_PATH + "last_execution.json", "r")
-        last_execution_details = json.loads(last_execution_file.read())
-
-        if last_execution_details != "Running":
-            last_execution_file = open(LOGS_PATH + "last_execution.json", "w")
-            last_execution_file.write(json.dumps("Running"))
-            last_execution_file.close()
-
-            flow_script.start_flow()
-
-            statistics = get_statistics()
+    current_time = datetime.now().ctime()
+    statistics = get_statistics()
 
-            last_execution_details = {"stats": statistics}
+    last_execution_details = {"executionTime": current_time, "stats": statistics}
+    last_ex_json = (json.dumps(last_execution_details))
 
-    except Exception as e:
-        last_execution_details = {"stats": {"Execution Error": str(e)}}
-        return abort(500)
-
-    finally:
-        current_time = datetime.now().ctime()
-
-        last_execution_details["executionTime"] = current_time
-        last_execution_file = open(LOGS_PATH + "last_execution.json", "w")
-        last_execution_file.write(json.dumps(last_execution_details))
-        last_execution_file.close()
+    # 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 = 'last_execution_time',
+            valcol = last_ex_json,
+            )
+        # If key already present in DB, do update instead
+        upsert = ins_stmt.on_conflict_do_update(
+                constraint='kv_unique_keycol_key',
+                set_=dict(valcol=last_ex_json)
+                )
+
+        try:
+            connection.execute(upsert)
+        except Exception as e:
+            current_app.logger.error("Insert/Update failed on Last Execution stats")
+            current_app.logger.exception(e)
 
     return jsonify(success=True)
 
 
 def get_statistics():
+
     with engine.connect() as connection:
         query_matches = text("SELECT count(*) FROM (SELECT distinct matching_id from pdp_contacts) as a;")
         query_total_count = text("SELECT count(*) FROM pdp_contacts;")
@@ -104,26 +112,22 @@ def get_statistics():
 
 @admin_api.route("/api/statistics", methods=["GET"])
 def list_statistics():
-    try:
-        last_execution_file = open(LOGS_PATH + "last_execution.json", "r")
-        last_execution_details = json.loads(last_execution_file.read())
-        last_execution_file.close()
+    """ Pull Last Execution stats from DB. """
+    current_app.logger.info("list_statistics() request")
+    last_execution_details = '{}'  # Empty but valid JSON
 
-    except (FileNotFoundError):
-        current_app.logger.error("last_execution.json file was missing")
-        return abort(500)
+    try:    # See Alembic Revision ID: 05e0693f8cbb for table definition
+        with engine.connect() as connection:
+            s = text("select valcol from kv_unique where keycol = 'last_execution_time';")
+            result = connection.execute(s)
+            last_execution_details  = result.fetchone()[0]
 
-    except (json.JSONDecodeError):
-        current_app.logger.error(
-            "last_execution.json could not be decoded - possible corruption"
-        )
-        return abort(500)
 
     except Exception as e:
-        current_app.logger.error("Failure reading last_execution.json: ", e)
-        return abort(500)
+        current_app.logger.error("Failure reading Last Execution stats from DB")
+       # return abort(500)    # Weird but not worth a 500
 
-    return jsonify(last_execution_details)
+    return last_execution_details
 
 
 """