Add function to delete phenotypes completely.

1 files changed, 55 insertions, 0 deletions

diff --git a/uploader/phenotypes/models.py b/uploader/phenotypes/models.py
index 3e1e31c..04abcc9 100644
--- a/uploader/phenotypes/models.py
+++ b/uploader/phenotypes/models.py
@@ -556,3 +556,58 @@ def delete_phenotypes_data(
                    data_ids)
     _ncount = cursor.rowcount
     return (_dcount, _secount, _ncount)
+
+
+def __linked_ids__(
+        cursor: BaseCursor,
+        population_id: int,
+        xref_ids: tuple[int, ...]
+) -> tuple[tuple[int, int, int], ...]:
+    """Retrieve `DataId` values from `PublishXRef` table."""
+    _paramstr = ", ".join(["%s"] * len(xref_ids))
+    cursor.execute("SELECT PhenotypeId, PublicationId, DataId "
+                   "FROM PublishXRef "
+                   f"WHERE InbredSetId=%s AND Id IN ({_paramstr})",
+                   (population_id,) + xref_ids)
+    return tuple(
+        (int(row["PhenotypeId"]), int(row["PublicationId"]), int(row["DataId"]))
+        for row in cursor.fetchall())
+
+
+def delete_phenotypes(
+        conn_or_cursor: Union[Connection, Cursor],
+        population_id: int,
+        xref_ids: tuple[int, ...]
+) -> tuple[int, int, int, int]:
+    """Delete phenotypes and all their data."""
+    def __delete_phenos__(cursor: BaseCursor, pheno_ids: tuple[int, ...]) -> int:
+        """Delete data from the `Phenotype` table."""
+        _paramstr = ", ".join(["%s"] * len(pheno_ids))
+        cursor.execute("DELETE FROM Phenotype "
+                       f"WHERE Id IN ({_paramstr})",
+                       pheno_ids)
+        return cursor.rowcount
+
+    def __delete_xrefs__(cursor: BaseCursor) -> int:
+        _paramstr = ", ".join(["%s"] * len(xref_ids))
+        cursor.execute("DELETE FROM PublishXRef "
+                       f"WHERE InbredSetId=%s AND Id IN ({_paramstr})",
+                       (population_id,) + xref_ids)
+        return cursor.rowcount
+
+    def __with_cursor__(cursor):
+        _phenoids, _pubids, _dataids = reduce(
+            lambda acc, curr: (acc[0] + (curr[0],),
+                               acc[1] + (curr[1],),
+                               acc[2] + (curr[2],)),
+            __linked_ids__(cursor, population_id, xref_ids),
+            (tuple(), tuple(), tuple()))
+        __delete_phenos__(cursor, _phenoids)
+        return (__delete_xrefs__(cursor),) + delete_phenotypes_data(
+            cursor, _dataids)
+
+    if isinstance(conn_or_cursor, BaseCursor):
+        return __with_cursor__(conn_or_cursor)
+
+    with conn_or_cursor.cursor(cursorclass=DictCursor) as cursor:
+        return __with_cursor__(cursor)