aboutsummaryrefslogtreecommitdiff
path: root/uploader/platforms/models.py
blob: 4b690bbd76002d14b0ef29610a43271efb241e78 (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
"""Handle db interactions for platforms."""
from typing import Optional

import MySQLdb as mdb
from MySQLdb.cursors import 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