Move the methods, "update" and "fetch", to gn3.db

1 files changed, 2 insertions, 62 deletions

diff --git a/gn3/db/phenotypes.py b/gn3/db/phenotypes.py
index ee523ad..2b93c85 100644
--- a/gn3/db/phenotypes.py
+++ b/gn3/db/phenotypes.py
@@ -1,17 +1,9 @@
 # pylint: disable=[R0902, R0903]
 """This contains all the necessary functions that access the phenotypes from
 the db"""
-from dataclasses import dataclass, asdict, astuple
+from dataclasses import dataclass
 
-from typing import Any, Dict, Optional
-from MySQLdb import escape_string
-
-from typing_extensions import Protocol
-
-
-class Dataclass(Protocol):
-    """Type Definition for a Dataclass"""
-    __dataclass_fields__: Dict
+from typing import Optional
 
 
 @dataclass(frozen=True)
@@ -107,55 +99,3 @@ publication_mapping = {
     "month": "Month",
     "year": "Year",
 }
-
-
-TABLEMAP = {
-    "Phenotype": phenotype_mapping,
-    "PublishXRef": publish_x_ref_mapping,
-    "Publication": publication_mapping,
-}
-
-DATACLASSMAP = {
-    "Phenotype": Phenotype,
-    "PublishXRef": PublishXRef,
-    "Publication": Publication,
-}
-
-
-def update(conn: Any,
-           table: str,
-           data: Dataclass,
-           where: Dataclass) -> Optional[int]:
-    """Run an UPDATE on a table"""
-    if not any(astuple(data) + astuple(where)):
-        return None
-    sql = f"UPDATE {table} SET "
-    sql += ", ".join(f"{TABLEMAP[table].get(k)} "
-                     f"= '{escape_string(str(v)).decode('utf-8')}'" for
-                     k, v in asdict(data).items()
-                     if v is not None and k in TABLEMAP[table])
-    sql += " WHERE "
-    sql += "AND ".join(f"{TABLEMAP[table].get(k)} = "
-                       f"'{escape_string(str(v)).decode('utf-8')}'" for
-                       k, v in asdict(where).items()
-                       if v is not None and k in TABLEMAP[table])
-    with conn.cursor() as cursor:
-        cursor.execute(sql)
-        return cursor.rowcount
-
-
-def fetchone(conn: Any,
-             table: str,
-             where: Dataclass) -> Optional[Dataclass]:
-    """Run a SELECT on a table. Returns only one result!"""
-    if not any(astuple(where)):
-        return None
-    sql = f"SELECT * FROM {table} "
-    sql += "WHERE "
-    sql += "AND ".join(f"{TABLEMAP[table].get(k)} = "
-                       f"'{escape_string(str(v)).decode('utf-8')}'" for
-                       k, v in asdict(where).items()
-                       if v is not None and k in TABLEMAP[table])
-    with conn.cursor() as cursor:
-        cursor.execute(sql)
-        return DATACLASSMAP[table](*cursor.fetchone())