aboutsummaryrefslogtreecommitdiff
path: root/gn3/db
diff options
context:
space:
mode:
Diffstat (limited to 'gn3/db')
-rw-r--r--gn3/db/__init__.py70
-rw-r--r--gn3/db/phenotypes.py64
2 files changed, 72 insertions, 62 deletions
diff --git a/gn3/db/__init__.py b/gn3/db/__init__.py
index e69de29..fae4d29 100644
--- a/gn3/db/__init__.py
+++ b/gn3/db/__init__.py
@@ -0,0 +1,70 @@
+# pylint: disable=[R0902, R0903]
+"""Module that exposes common db operations"""
+from typing import Optional, Dict, Any
+from dataclasses import dataclass, asdict, astuple
+from typing_extensions import Protocol
+from MySQLdb import escape_string
+
+from gn3.db.phenotypes import Phenotype
+from gn3.db.phenotypes import PublishXRef
+from gn3.db.phenotypes import Publication
+
+from gn3.db.phenotypes import phenotype_mapping
+from gn3.db.phenotypes import publish_x_ref_mapping
+from gn3.db.phenotypes import publication_mapping
+
+TABLEMAP = {
+ "Phenotype": phenotype_mapping,
+ "PublishXRef": publish_x_ref_mapping,
+ "Publication": publication_mapping,
+}
+
+DATACLASSMAP = {
+ "Phenotype": Phenotype,
+ "PublishXRef": PublishXRef,
+ "Publication": Publication,
+}
+
+
+class Dataclass(Protocol):
+ """Type Definition for a Dataclass"""
+ __dataclass_fields__: Dict
+
+
+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())
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())