1 files changed, 95 insertions, 0 deletions
diff --git a/uploader/platforms/models.py b/uploader/platforms/models.py
new file mode 100644
index 0000000..a859371
--- /dev/null
+++ b/uploader/platforms/models.py
@@ -0,0 +1,95 @@
+"""Handle db interactions for platforms."""
+from typing import Optional
+
+import MySQLdb as mdb
+from MySQLdb.cursors import Cursor, DictCursor
+
+def platforms_by_species(
+        conn: mdb.Connection,
+        speciesid: int,
+        offset: int = 0,
+        limit: Optional[int] = None
+) -> tuple[dict, ...]:
+    """Retrieve platforms by the species"""
+    _query = ("SELECT * FROM GeneChip WHERE SpeciesId=%s "
+              "ORDER BY GeneChipName ASC")
+    if bool(limit) and limit > 0:# type: ignore[operator]
+        _query = f"{_query} LIMIT {limit} OFFSET {offset}"
+    with conn.cursor(cursorclass=DictCursor) as cursor:
+        cursor.execute(_query, (speciesid,))
+        return tuple(dict(row) for row in cursor.fetchall())
+
+
+def species_platforms_count(conn: mdb.Connection, species_id: int) -> int:
+    """Get the number of platforms in the database for a particular species."""
+    with conn.cursor(cursorclass=DictCursor) as cursor:
+        cursor.execute(
+            "SELECT COUNT(GeneChipName) AS count FROM GeneChip "
+            "WHERE SpeciesId=%s",
+            (species_id,))
+        return int(cursor.fetchone()["count"])
+
+
+def platform_by_id(conn: mdb.Connection, platformid: int) -> Optional[dict]:
+    """Retrieve a platform by its ID"""
+    with conn.cursor(cursorclass=DictCursor) as cursor:
+        cursor.execute("SELECT * FROM GeneChip WHERE Id=%s",
+                       (platformid,))
+        result = cursor.fetchone()
+        if bool(result):
+            return dict(result)
+
+    return None
+
+
+def platform_by_species_and_id(
+        conn: mdb.Connection, species_id: int, platformid: int
+) -> Optional[dict]:
+    """Retrieve a platform by its species and ID"""
+    with conn.cursor(cursorclass=DictCursor) as cursor:
+        cursor.execute("SELECT * FROM GeneChip WHERE SpeciesId=%s AND Id=%s",
+                       (species_id, platformid))
+        result = cursor.fetchone()#pylint: disable=[duplicate-code]
+        if bool(result):
+            return dict(result)
+
+    return None
+
+
+def save_new_platform(# pylint: disable=[too-many-arguments]
+        cursor: Cursor,
+        species_id: int,
+        geo_platform: str,
+        platform_name: str,
+        platform_shortname: str,
+        platform_title: str,
+        go_tree_value: Optional[str]
+) -> dict:
+    """Save a new platform to the database."""
+    params = {
+        "species_id": species_id,
+        "GeoPlatform": geo_platform,
+        "GeneChipName": platform_name,
+        "Name": platform_shortname,
+        "Title": platform_title,
+        "GO_tree_value": go_tree_value
+    }
+    cursor.execute("SELECT SpeciesId, GeoPlatform FROM GeneChip")
+    assert (species_id, geo_platform) not in (
+        (row["SpeciesId"], row["GeoPlatform"]) for row in cursor.fetchall())
+    cursor.execute(
+        "INSERT INTO "
+        "GeneChip(SpeciesId, GeneChipName, Name, GeoPlatform, Title, GO_tree_value) "
+        "VALUES("
+        "%(species_id)s, %(GeneChipName)s, %(Name)s, %(GeoPlatform)s, "
+        "%(Title)s, %(GO_tree_value)s"
+        ")",
+        params)
+    new_id = cursor.lastrowid
+    cursor.execute("UPDATE GeneChip SET GeneChipId=%s WHERE Id=%s",
+                   (new_id, new_id))
+    return {
+        **params,
+        "Id": new_id,
+        "GeneChipId": new_id
+    }