Fetch `metadata_audit` trail with direct query functions.

1 files changed, 38 insertions, 0 deletions

diff --git a/gn3/db/metadata_audit.py b/gn3/db/metadata_audit.py
index 707892b..1b6e8a9 100644
--- a/gn3/db/metadata_audit.py
+++ b/gn3/db/metadata_audit.py
@@ -3,6 +3,7 @@
 table from the db
 
 """
+import json
 from typing import Optional
 from dataclasses import dataclass
 
@@ -36,3 +37,40 @@ def create_metadata_audit(conn, metadata: dict) -> int:
             "VALUES (%(dataset_id)s, %(editor)s, %(json_data)s)",
             metadata)
         return cursor.rowcount
+
+def __parse_metadata_audit__(row) -> dict:
+    """Convert values in DB to expected Python values """
+    return {
+        **{key:val for key,val in row.items() if key not in ("json_diff_data")},
+        "id_": row["id"],
+        "json_data": json.loads(row["json_diff_data"])
+    }
+
+def fetch_phenotype_metadata_audit_by_dataset_id(conn, dataset_id) -> tuple[dict, ...]:
+    """Fetch phenotype a metadata audit trail by `dataset_id`."""
+    assert bool(dataset_id), "`dataset_id` MUST be provided."
+    with conn.cursor(cursorclass=DictCursor) as cursor:
+        cursor.execute(
+            "SELECT ma.* "
+            "FROM PublishXRef AS pxr LEFT JOIN metadata_audit AS ma "
+            "ON pxr.Id=ma.dataset_id "
+            "WHERE pxr.Id=%(dataset_id)s "
+            "AND ma.json_diff_data LIKE '%%phenotype%%' "
+            "ORDER BY time_stamp ASC",
+            {"dataset_id": dataset_id})
+        __pk__(cursor._executed, "THE RUN QUERY")
+        return tuple(__parse_metadata_audit__(row) for row in cursor.fetchall())
+
+def fetch_probeset_metadata_audit_by_trait_name(conn, trait_name) -> tuple[dict, ...]:
+    """Fetch a probeset metadata audit trail by `dataset_id`."""
+    assert bool(dataset_id), "`dataset_id` MUST be provided."
+    with conn.cursor(cursorclass=DictCursor) as cursor:
+        cursor.execute(
+            "SELECT ma.* "
+            "FROM ProbeSet AS ps LEFT JOIN metadata_audit AS ma "
+            "ON ps.Id=ma.dataset_id "
+            "WHERE ps.Name=%(trait_name)s "
+            "AND json_diff_data LIKE '%probeset%' "
+            "ORDER BY time_stamp ASCENDING",
+            {"trait_name": trait_name})
+        return tuple(__parse_metadata_audit__(row) for row in cursor.fetchall())