aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--gn3/db/metadata_audit.py38
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())