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
96
97
98
99
100
101
102
103
104
105
106
|
"""Handles the resource objects' data."""
from MySQLdb.cursors import DictCursor
from gn_auth.auth.db import mariadb as gn3db
from gn_auth.auth.db import sqlite3 as authdb
from gn_auth.auth.authorisation.groups import Group
from gn_auth.auth.authorisation.checks import authorised_p
from gn_auth.auth.authorisation.errors import NotFoundError
def __fetch_mrna_data_by_ids__(
conn: gn3db.DbConnection, dataset_ids: tuple[str, ...]) -> tuple[
dict, ...]:
"""Fetch mRNA Assay data by ID."""
with conn.cursor(DictCursor) as cursor:
paramstr = ", ".join(["%s"] * len(dataset_ids))
cursor.execute(
"SELECT psf.Id, psf.Name AS dataset_name, "
"psf.FullName AS dataset_fullname, "
"ifiles.GN_AccesionId AS accession_id FROM ProbeSetFreeze AS psf "
"INNER JOIN InfoFiles AS ifiles ON psf.Name=ifiles.InfoPageName "
f"WHERE psf.Id IN ({paramstr})",
dataset_ids)
res = cursor.fetchall()
if res:
return tuple(dict(row) for row in res)
raise NotFoundError("Could not find mRNA Assay data with the given ID.")
def __fetch_geno_data_by_ids__(
conn: gn3db.DbConnection, dataset_ids: tuple[str, ...]) -> tuple[
dict, ...]:
"""Fetch genotype data by ID."""
with conn.cursor(DictCursor) as cursor:
paramstr = ", ".join(["%s"] * len(dataset_ids))
cursor.execute(
"SELECT gf.Id, gf.Name AS dataset_name, "
"gf.FullName AS dataset_fullname, "
"ifiles.GN_AccesionId AS accession_id FROM GenoFreeze AS gf "
"INNER JOIN InfoFiles AS ifiles ON gf.Name=ifiles.InfoPageName "
f"WHERE gf.Id IN ({paramstr})",
dataset_ids)
res = cursor.fetchall()
if res:
return tuple(dict(row) for row in res)
raise NotFoundError("Could not find Genotype data with the given ID.")
def __fetch_pheno_data_by_ids__(
conn: gn3db.DbConnection, dataset_ids: tuple[str, ...]) -> tuple[
dict, ...]:
"""Fetch phenotype data by ID."""
with conn.cursor(DictCursor) as cursor:
paramstr = ", ".join(["%s"] * len(dataset_ids))
cursor.execute(
"SELECT pxf.Id, iset.InbredSetName, pf.Id AS dataset_id, "
"pf.Name AS dataset_name, pf.FullName AS dataset_fullname, "
"ifiles.GN_AccesionId AS accession_id "
"FROM PublishXRef AS pxf "
"INNER JOIN InbredSet AS iset ON pxf.InbredSetId=iset.InbredSetId "
"INNER JOIN PublishFreeze AS pf ON iset.InbredSetId=pf.InbredSetId "
"INNER JOIN InfoFiles AS ifiles ON pf.Name=ifiles.InfoPageName "
f"WHERE pxf.Id IN ({paramstr})",
dataset_ids)
res = cursor.fetchall()
if res:
return tuple(dict(row) for row in res)
raise NotFoundError(
"Could not find Phenotype/Publish data with the given IDs.")
def __fetch_data_by_id(
conn: gn3db.DbConnection, dataset_type: str,
dataset_ids: tuple[str, ...]) -> tuple[dict, ...]:
"""Fetch data from MySQL by IDs."""
fetch_fns = {
"mrna": __fetch_mrna_data_by_ids__,
"genotype": __fetch_geno_data_by_ids__,
"phenotype": __fetch_pheno_data_by_ids__
}
return fetch_fns[dataset_type](conn, dataset_ids)
@authorised_p(("system:data:link-to-group",),
error_description=(
"You do not have sufficient privileges to link data to (a) "
"group(s)."),
oauth2_scope="profile group resource")
def link_data_to_group(
authconn: authdb.DbConnection, gn3conn: gn3db.DbConnection,
dataset_type: str, dataset_ids: tuple[str, ...], group: Group) -> tuple[
dict, ...]:
"""Link the given data to the specified group."""
the_data = __fetch_data_by_id(gn3conn, dataset_type, dataset_ids)
with authdb.cursor(authconn) as cursor:
params = tuple({
"group_id": str(group.group_id), "dataset_type": {
"mrna": "mRNA", "genotype": "Genotype",
"phenotype": "Phenotype"
}[dataset_type],
"dataset_or_trait_id": item["Id"],
"dataset_name": item["dataset_name"],
"dataset_fullname": item["dataset_fullname"],
"accession_id": item["accession_id"]
} for item in the_data)
cursor.executemany(
"INSERT INTO linked_group_data VALUES"
"(:group_id, :dataset_type, :dataset_or_trait_id, :dataset_name, "
":dataset_fullname, :accession_id)",
params)
return params
|
