aboutsummaryrefslogtreecommitdiff
path: root/uploader/db/tissues.py
blob: 9fe7bab61dcf9c510b68e65896a4c6b46688efc6 (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
"""Handle db interactions for tissue."""
from typing import Union, Optional

import MySQLdb as mdb
from MySQLdb.cursors import DictCursor

def all_tissues(conn: mdb.Connection) -> tuple[dict, ...]:
    """All available tissue."""
    with conn.cursor(cursorclass=DictCursor) as cursor:
        cursor.execute("SELECT * FROM Tissue ORDER BY TissueName")
        return tuple(dict(row) for row in cursor.fetchall())


def tissue_by_id(conn: mdb.Connection, tissueid) -> Optional[dict]:
    """Retrieve a tissue by its ID"""
    with conn.cursor(cursorclass=DictCursor) as cursor:
        cursor.execute("SELECT * FROM Tissue WHERE Id=%s", (tissueid,))
        result = cursor.fetchone()
        if bool(result):
            return dict(result)

    return None


def create_new_tissue(
        conn: mdb.Connection,
        name: str,
        shortname: str,
        birnlexid: Optional[str] = None,
        birnlexname: Optional[str] = None
) -> dict[str, Union[int, str, None]]:
    """Add a new tissue, organ or biological material to the database."""
    with conn.cursor() as cursor:
        cursor.execute(
            "INSERT INTO "
            "Tissue(TissueName, Name, Short_Name, BIRN_lex_ID, BIRN_lex_Name) "
            "VALUES (%s, %s, %s, %s, %s)",
            (name, name, shortname, birnlexid, birnlexname))
        tissueid = cursor.lastrowid
        cursor.execute("UPDATE Tissue SET TissueId=%s WHERE Id=%s",
                       (tissueid, tissueid))
        return {
            "Id": tissueid,
            "TissueId": tissueid,
            "TissueName": name,
            "Name": name,
            "Short_Name": shortname,
            "BIRN_lex_ID": birnlexid,
            "BIRN_lex_Name": birnlexname
        }