diff options
Diffstat (limited to 'uploader/db')
| -rw-r--r-- | uploader/db/__init__.py | 2 | ||||
| -rw-r--r-- | uploader/db/averaging.py | 23 | ||||
| -rw-r--r-- | uploader/db/datasets.py | 133 | ||||
| -rw-r--r-- | uploader/db/platforms.py | 25 | ||||
| -rw-r--r-- | uploader/db/tissues.py | 50 |
5 files changed, 233 insertions, 0 deletions
diff --git a/uploader/db/__init__.py b/uploader/db/__init__.py new file mode 100644 index 0000000..d2b1d9d --- /dev/null +++ b/uploader/db/__init__.py @@ -0,0 +1,2 @@ +"""Database functions""" +from .datasets import geno_datasets_by_species_and_population diff --git a/uploader/db/averaging.py b/uploader/db/averaging.py new file mode 100644 index 0000000..62bbe67 --- /dev/null +++ b/uploader/db/averaging.py @@ -0,0 +1,23 @@ +"""Functions for db interactions for averaging methods""" +from typing import Optional + +import MySQLdb as mdb +from MySQLdb.cursors import DictCursor + +def averaging_methods(conn: mdb.Connection) -> tuple[dict, ...]: + """Fetch all available averaging methods""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM AvgMethod") + return tuple(dict(row) for row in cursor.fetchall()) + +def averaging_method_by_id( + conn: mdb.Connection, averageid: int) -> Optional[dict]: + """Fetch the averaging method by its ID""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM AvgMethod WHERE Id=%s", + (averageid,)) + result = cursor.fetchone() + if bool(result): + return dict(result) + + return None diff --git a/uploader/db/datasets.py b/uploader/db/datasets.py new file mode 100644 index 0000000..767ec41 --- /dev/null +++ b/uploader/db/datasets.py @@ -0,0 +1,133 @@ +"""Functions for accessing the database relating to datasets.""" +from datetime import date +from typing import Optional + +import MySQLdb as mdb +from MySQLdb.cursors import DictCursor + +def geno_datasets_by_species_and_population( + conn: mdb.Connection, + speciesid: int, + populationid: int) -> tuple[dict, ...]: + """Retrieve all genotypes datasets by species and population""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute( + "SELECT gf.* FROM InbredSet AS iset INNER JOIN GenoFreeze AS gf " + "ON iset.InbredSetId=gf.InbredSetId " + "WHERE iset.SpeciesId=%(sid)s AND iset.InbredSetId=%(pid)s", + {"sid": speciesid, "pid": populationid}) + return tuple(dict(row) for row in cursor.fetchall()) + +def geno_dataset_by_id(conn: mdb.Connection, dataset_id) -> Optional[dict]: + """Retrieve genotype dataset by ID""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM GenoFreeze WHERE Id=%s", (dataset_id,)) + _dataset = cursor.fetchone() + return dict(_dataset) if bool(_dataset) else None + +def probeset_studies_by_species_and_population( + conn: mdb.Connection, + speciesid: int, + populationid: int) -> tuple[dict, ...]: + """Retrieve all probesets""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute( + "SELECT pf.* FROM InbredSet AS iset INNER JOIN ProbeFreeze AS pf " + "ON iset.InbredSetId=pf.InbredSetId " + "WHERE iset.SpeciesId=%(sid)s AND iset.InbredSetId=%(pid)s", + {"sid": speciesid, "pid": populationid}) + return tuple(dict(row) for row in cursor.fetchall()) + +def probeset_datasets_by_study(conn: mdb.Connection, + studyid: int) -> tuple[dict, ...]: + """Retrieve all probeset databases by study.""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM ProbeSetFreeze WHERE ProbeFreezeId=%s", + (studyid,)) + return tuple(dict(row) for row in cursor.fetchall()) + +def probeset_study_by_id(conn: mdb.Connection, studyid) -> Optional[dict]: + """Retrieve ProbeSet study by ID""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM ProbeFreeze WHERE Id=%s", (studyid,)) + _study = cursor.fetchone() + return dict(_study) if bool(_study) else None + +def probeset_create_study(conn: mdb.Connection,#pylint: disable=[too-many-arguments] + populationid: int, + platformid: int, + tissueid: int, + studyname: str, + studyfullname: str = "", + studyshortname: str = ""): + """Create a new ProbeSet study.""" + with conn.cursor(cursorclass=DictCursor) as cursor: + studydata = { + "platid": platformid, + "tissueid": tissueid, + "name": studyname, + "fname": studyfullname or studyname, + "sname": studyshortname, + "today": date.today().isoformat(), + "popid": populationid + } + cursor.execute( + """ + INSERT INTO ProbeFreeze( + ChipId, TissueId, Name, FullName, ShortName, CreateTime, + InbredSetId + ) VALUES ( + %(platid)s, %(tissueid)s, %(name)s, %(fname)s, %(sname)s, + %(today)s, %(popid)s + ) + """, + studydata) + studyid = cursor.lastrowid + cursor.execute("UPDATE ProbeFreeze SET ProbeFreezeId=%s WHERE Id=%s", + (studyid, studyid)) + return {**studydata, "studyid": studyid} + +def probeset_create_dataset(conn: mdb.Connection,#pylint: disable=[too-many-arguments] + studyid: int, + averageid: int, + datasetname: str, + datasetfullname: str, + datasetshortname: str="", + public: bool = True, + datascale="log2") -> dict: + """Create a new ProbeSet dataset.""" + with conn.cursor(cursorclass=DictCursor) as cursor: + dataset = { + "studyid": studyid, + "averageid": averageid, + "name2": datasetname, + "fname": datasetfullname, + "name": datasetshortname, + "sname": datasetshortname, + "today": date.today().isoformat(), + "public": 2 if public else 0, + "authorisedusers": "williamslab", + "datascale": datascale + } + cursor.execute( + """ + INSERT INTO ProbeSetFreeze( + ProbeFreezeId, AvgId, Name, Name2, FullName, ShortName, + CreateTime, public, AuthorisedUsers, DataScale) + VALUES( + %(studyid)s, %(averageid)s, %(name)s, %(name2)s, %(fname)s, + %(sname)s, %(today)s, %(public)s, %(authorisedusers)s, + %(datascale)s) + """, + dataset) + return {**dataset, "datasetid": cursor.lastrowid} + +def probeset_dataset_by_id(conn: mdb.Connection, datasetid) -> Optional[dict]: + """Fetch a ProbeSet dataset by its ID""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM ProbeSetFreeze WHERE Id=%s", (datasetid,)) + result = cursor.fetchone() + if bool(result): + return dict(result) + + return None diff --git a/uploader/db/platforms.py b/uploader/db/platforms.py new file mode 100644 index 0000000..cb527a7 --- /dev/null +++ b/uploader/db/platforms.py @@ -0,0 +1,25 @@ +"""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) -> tuple[dict, ...]: + """Retrieve platforms by the species""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM GeneChip WHERE SpeciesId=%s " + "ORDER BY GeneChipName ASC", + (speciesid,)) + return tuple(dict(row) for row in cursor.fetchall()) + +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 diff --git a/uploader/db/tissues.py b/uploader/db/tissues.py new file mode 100644 index 0000000..9fe7bab --- /dev/null +++ b/uploader/db/tissues.py @@ -0,0 +1,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 + } |