Merge pull request #29 from DataKitchen/release/3.1.2

Release/3.1.2

Author: datakitchen-devops

pyproject.toml

@@ -8,7 +8,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dataops-testgen"
-version = "3.1.0"
+version = "3.1.2"
 description = "DataKitchen's Data Quality DataOps TestGen"
 authors = [
     { "name" = "DataKitchen, Inc.", "email" = "info@datakitchen.io" },

testgen/__main__.py

@@ -43,6 +43,7 @@
     logs,
     version_service,
 )
+from testgen.ui.queries import profiling_run_queries, test_run_queries
 from testgen.utils import plugins
 
 LOG = logging.getLogger("testgen")
@@ -606,6 +607,11 @@ def run(debug: bool):
     use_ssl = os.path.isfile(settings.SSL_CERT_FILE) and os.path.isfile(settings.SSL_KEY_FILE)
 
     patch_streamlit.patch(force=True)
+    try:
+        profiling_run_queries.cancel_all_running()
+        test_run_queries.cancel_all_running()
+    except Exception:
+        LOG.warning("Failed to cancel 'Running' profiling/test runs")
 
     try:
         app_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ui/app.py")

testgen/commands/run_profiling_bridge.py

@@ -17,8 +17,6 @@
     RunActionQueryList,
     RunThreadedRetrievalQueryList,
     WriteListToDB,
-    date_service,
-    read_template_sql_file,
 )
 from testgen.common.database.database_service import empty_cache
 
@@ -508,14 +506,3 @@ def run_profiling_queries(strTableGroupsID, spinner=None):
             str_error_status = "successfully."
         message += str_error_status
     return message
-
-
-def update_profile_run_status(profile_run_id, status):
-    sql_template = read_template_sql_file("project_profile_run_record_update_status.sql", sub_directory="profiling")
-
-    sql_template = sql_template.replace("{STATUS}", status)
-    sql_template = sql_template.replace("{NOW}", date_service.get_now_as_string())
-    sql_template = sql_template.replace("{EXCEPTION_MESSAGE}", "")
-    sql_template = sql_template.replace("{PROFILE_RUN_ID}", profile_run_id)
-
-    RunActionQueryList("DKTG", [sql_template])

testgen/template/profiling/project_profile_run_record_update_status.sql

@@ -17,7 +17,7 @@ anomalies AS (
         EXTRACT(
             EPOCH
             FROM runs.profiling_starttime
-        ) AS time,
+        ) * 1000 AS time,
         '' AS name,
         runs.id::text AS run_id,
         'hygiene' AS issue_type
@@ -50,7 +50,7 @@ tests AS (
         EXTRACT(
             EPOCH
             FROM test_time
-        ) AS time,
+        ) * 1000 AS time,
         test_suites.test_suite AS name,
         test_results.test_run_id::text AS run_id,
         'test' AS issue_type
@@ -80,4 +80,4 @@ ORDER BY
         WHEN 'Possible' THEN 4
         WHEN 'Warning' THEN 5
         ELSE 6
-    END
+    END

testgen/template/score_cards/get_score_card_issues_by_column.sql

@@ -17,7 +17,7 @@ anomalies AS (
         EXTRACT(
             EPOCH
             FROM runs.profiling_starttime
-        ) AS time,
+        ) * 1000 AS time,
         '' AS name,
         runs.id::text AS run_id,
         'hygiene' AS issue_type
@@ -51,7 +51,7 @@ tests AS (
         EXTRACT(
             EPOCH
             FROM test_time
-        ) AS time,
+        ) * 1000 AS time,
         test_suites.test_suite AS name,
         test_results.test_run_id::text AS run_id,
         'test' AS issue_type
@@ -82,4 +82,4 @@ ORDER BY
         WHEN 'Possible' THEN 4
         WHEN 'Warning' THEN 5
         ELSE 6
-    END
+    END

testgen/template/score_cards/get_score_card_issues_by_dimension.sql

@@ -0,0 +1,27 @@
+import streamlit as st
+
+import testgen.ui.services.database_service as db
+from testgen.common import date_service
+
+
+def update_status(profile_run_id: str, status: str) -> None:
+    schema: str = st.session_state["dbschema"]
+    now = date_service.get_now_as_string()
+
+    sql = f"""
+    UPDATE {schema}.profiling_runs
+    SET status = '{status}',
+        profiling_endtime = '{now}'
+    WHERE id = '{profile_run_id}'::UUID;
+    """
+    db.execute_sql(sql)
+    st.cache_data.clear()
+
+
+def cancel_all_running() -> None:
+    schema: str = db.get_schema()
+    db.execute_sql(f"""
+    UPDATE {schema}.profiling_runs
+        SET status = 'Cancelled'
+        WHERE status = 'Running';
+    """)

testgen/ui/queries/profiling_run_queries.py

@@ -4,10 +4,11 @@
 import testgen.ui.services.database_service as db
 
 
-def cascade_delete(schema: str, test_suite_ids: list[str]) -> None:
+def cascade_delete(test_suite_ids: list[str]) -> None:
     if not test_suite_ids:
         raise ValueError("No Test Suite is specified.")
 
+    schema: str = st.session_state["dbschema"]
     ids_str = ", ".join([f"'{item}'" for item in test_suite_ids])
     sql = f"""
         DELETE
@@ -23,10 +24,11 @@ def cascade_delete(schema: str, test_suite_ids: list[str]) -> None:
     st.cache_data.clear()
 
 
-def update_status(schema: str, test_run_id: str, status: str) -> None:
+def update_status(test_run_id: str, status: str) -> None:
     if not all([test_run_id, status]):
         raise ValueError("Missing query parameters.")
 
+    schema: str = st.session_state["dbschema"]
     now = date_service.get_now_as_string()
 
     sql = f"""
@@ -37,3 +39,12 @@ def update_status(schema: str, test_run_id: str, status: str) -> None:
     """
     db.execute_sql(sql)
     st.cache_data.clear()
+
+
+def cancel_all_running() -> None:
+    schema: str = db.get_schema()
+    db.execute_sql(f"""
+    UPDATE {schema}.test_runs
+        SET status = 'Cancelled'
+        WHERE status = 'Running';
+    """)

testgen/ui/queries/test_run_queries.py

@@ -4,7 +4,7 @@
 import testgen.ui.services.connection_service as connection_service
 import testgen.ui.services.database_service as database_service
 import testgen.ui.services.table_group_service as table_group_service
-import testgen.ui.services.test_run_service as test_run_service
+from testgen.ui.queries import test_run_queries
 
 
 def update_attribute(test_definition_ids, attribute, value):
@@ -54,7 +54,7 @@ def delete(test_definition_ids, dry_run=False):
 
 def cascade_delete(test_suite_ids: list[str]):
     schema = st.session_state["dbschema"]
-    test_run_service.cascade_delete(test_suite_ids)
+    test_run_queries.cascade_delete(test_suite_ids)
     test_definition_queries.cascade_delete(schema, test_suite_ids)
 
 
@@ -138,7 +138,7 @@ def validate_test(test_definition):
     )
 
 
-def move(test_definitions, target_table_group, target_test_suite):    
+def move(test_definitions, target_table_group, target_test_suite):
     schema = st.session_state["dbschema"]
     test_definition_queries.move(schema, test_definitions, target_table_group, target_test_suite)
 
@@ -152,4 +152,3 @@ def copy(test_definitions, target_table_group, target_test_suite):
 def get_test_definitions_collision(test_definitions, target_table_group, target_test_suite):
     schema = st.session_state["dbschema"]
     return test_definition_queries.get_test_definitions_collision(schema, test_definitions, target_table_group, target_test_suite)
-