aboutsummaryrefslogtreecommitdiff
path: root/uploader/platforms/models.py
blob: a859371543b706219124f1e999c6e63cf4808618 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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
    }