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
|
"""Functions for accessing the database relating to datasets."""
from datetime import date
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: int) -> dict:
"""Retrieve genotype dataset by ID"""
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute("SELECT * FROM GenoFreeze WHERE Id=%s",
(dataset_id,))
return dict(cursor.fetchone())
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: int) -> dict:
"""Retrieve ProbeSet study by ID"""
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute("SELECT * FROM ProbeFreeze WHERE Id=%s",
(studyid,))
return dict(cursor.fetchone())
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",
(studyid,))
return {**studydata, "studyid": studyid}
|
