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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
|
"""Handles the resource objects' data."""
from typing import Any, Sequence
from MySQLdb.cursors import DictCursor
from gn3 import db_utils as gn3db
from gn3.auth import db as authdb
from gn3.auth.authorisation.groups import Group
from gn3.auth.authorisation.checks import authorised_p
from gn3.auth.authorisation.errors import InvalidData, NotFoundError
def __fetch_grouped_data__(
conn: authdb.DbConnection, dataset_type: str) -> Sequence[dict[str, Any]]:
"""Retrieve ids for all data that are linked to groups in the auth db."""
with authdb.cursor(conn) as cursor:
cursor.execute(
"SELECT dataset_type, dataset_or_trait_id FROM linked_group_data "
"WHERE LOWER(dataset_type)=?",
(dataset_type,))
return tuple(dict(row) for row in cursor.fetchall())
def __fetch_ungrouped_mrna_data__(
conn: gn3db.Connection, grouped_data, offset: int) -> Sequence[dict]:
"""Fetch ungrouped mRNA Assay data."""
query = ("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")
params: tuple[Any, ...] = tuple()
if bool(grouped_data):
clause = ", ".join(["%s"] * len(grouped_data))
query = f"{query} WHERE psf.Id NOT IN ({clause})"
params = tuple(item["dataset_or_trait_id"] for item in grouped_data)
query = f"{query} LIMIT 100 OFFSET %s"
with conn.cursor(DictCursor) as cursor:
cursor.execute(query, (params + (offset,)))
return tuple(dict(row) for row in cursor.fetchall())
def __fetch_ungrouped_geno_data__(
conn: gn3db.Connection, grouped_data, offset: int) -> Sequence[dict]:
"""Fetch ungrouped Genotype data."""
query = ("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")
params: tuple[Any, ...] = tuple()
if bool(grouped_data):
clause = ", ".join(["%s"] * len(grouped_data))
query = f"{query} WHERE gf.Id NOT IN ({clause})"
params = tuple(item["dataset_or_trait_id"] for item in grouped_data)
query = f"{query} LIMIT 100 OFFSET %s"
with conn.cursor(DictCursor) as cursor:
cursor.execute(query, (params + (offset,)))
return tuple(dict(row) for row in cursor.fetchall())
def __fetch_ungrouped_pheno_data__(
conn: gn3db.Connection, grouped_data, offset: int) -> Sequence[dict]:
"""Fetch ungrouped Phenotype data."""
query = ("SELECT "
"pxf.Id, iset.InbredSetName, pf.Name AS dataset_name, "
"pf.FullName AS dataset_fullname, "
"pf.ShortName AS dataset_shortname "
"FROM PublishXRef AS pxf "
"INNER JOIN InbredSet AS iset "
"ON pxf.InbredSetId=iset.InbredSetId "
"LEFT JOIN PublishFreeze AS pf "
"ON iset.InbredSetId=pf.InbredSetId")
params: tuple[Any, ...] = tuple()
if bool(grouped_data):
clause = ", ".join(["%s"] * len(grouped_data))
query = f"{query} WHERE pxf.Id NOT IN ({clause})"
params = tuple(item["dataset_or_trait_id"] for item in grouped_data)
query = f"{query} LIMIT 100 OFFSET %s"
with conn.cursor(DictCursor) as cursor:
cursor.execute(query, (params + (offset,)))
return tuple(dict(row) for row in cursor.fetchall())
def __fetch_ungrouped_data__(
conn: gn3db.Connection, dataset_type: str,
ungrouped: Sequence[dict[str, Any]],
offset) -> Sequence[dict[str, Any]]:
"""Fetch any ungrouped data."""
fetch_fns = {
"mrna": __fetch_ungrouped_mrna_data__,
"genotype": __fetch_ungrouped_geno_data__,
"phenotype": __fetch_ungrouped_pheno_data__
}
return fetch_fns[dataset_type](conn, ungrouped, offset)
@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 retrieve_ungrouped_data(
authconn: authdb.DbConnection,
gn3conn: gn3db.Connection,
dataset_type: str,
offset: int = 0) -> Sequence[dict]:
"""Retrieve any data not linked to any group."""
if dataset_type not in ("mrna", "genotype", "phenotype"):
raise InvalidData(
"Requested dataset type is invalid. Expected one of "
"'mrna', 'genotype' or 'phenotype'.")
grouped_data = __fetch_grouped_data__(authconn, dataset_type)
return __fetch_ungrouped_data__(gn3conn, dataset_type, grouped_data, offset)
def __fetch_mrna_data_by_id__(conn: gn3db.Connection, dataset_id: str) -> dict:
"""Fetch mRNA Assay data by ID."""
with conn.cursor(DictCursor) as cursor:
cursor.execute(
"SELECT psf.Id, psf.Name, psf.FullName, "
"ifiles.GN_AccesionId AS accession_id FROM ProbeSetFreeze AS psf "
"INNER JOIN InfoFiles AS ifiles ON psf.Name=ifiles.InfoPageName "
"WHERE psf.Id=%s",
(dataset_id,))
res = cursor.fetchone()
if res:
return dict(res)
raise NotFoundError("Could not find mRNA Assay data with the given ID.")
def __fetch_geno_data_by_id__(conn: gn3db.Connection, dataset_id: str) -> dict:
"""Fetch genotype data by ID."""
with conn.cursor(DictCursor) as cursor:
cursor.execute(
"SELECT gf.Id, gf.Name, gf.FullName, "
"ifiles.GN_AccesionId AS accession_id FROM GenoFreeze AS gf "
"INNER JOIN InfoFiles AS ifiles ON gf.Name=ifiles.InfoPageName "
"WHERE gf.Id=%s",
(dataset_id,))
res = cursor.fetchone()
if res:
return dict(res)
raise NotFoundError("Could not find Genotype data with the given ID.")
def __fetch_pheno_data_by_id__(conn: gn3db.Connection, dataset_id: str) -> dict:
"""Fetch phenotype data by ID."""
with conn.cursor(DictCursor) as cursor:
cursor.execute(
"SELECT pf.Id, pf.Name, pf.FullName, "
"ifiles.GN_AccesionId AS accession_id FROM PublishFreeze AS pf "
"INNER JOIN InfoFiles AS ifiles ON pf.Name=ifiles.InfoPageName "
"WHERE pf.Id=%s",
(dataset_id,))
res = cursor.fetchone()
if res:
return dict(res)
raise NotFoundError(
"Could not find Phenotype/Publish data with the given ID.")
def __fetch_data_by_id(
conn: gn3db.Connection, dataset_type: str, dataset_id: str) -> dict:
"""Fetch data from MySQL by ID."""
fetch_fns = {
"mrna": __fetch_mrna_data_by_id__,
"genotype": __fetch_geno_data_by_id__,
"phenotype": __fetch_pheno_data_by_id__
}
return fetch_fns[dataset_type](conn, dataset_id)
@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.Connection,
dataset_type: str, dataset_id: str, group: Group) -> dict:
"""Link the given data to the specified group."""
the_data = __fetch_data_by_id(gn3conn, dataset_type, dataset_id)
with authdb.cursor(authconn) as cursor:
params = {
"group_id": str(group.group_id), "dataset_type": {
"mrna": "mRNA", "genotype": "Genotype",
"phenotype": "Phenotype"
}[dataset_type],
"dataset_or_trait_id": dataset_id, "dataset_name": the_data["Name"],
"dataset_fullname": the_data["FullName"],
"accession_id": the_data["accession_id"]
}
cursor.execute(
"INSERT INTO linked_group_data VALUES"
"(:group_id, :dataset_type, :dataset_or_trait_id, :dataset_name, "
":dataset_fullname, :accession_id)",
params)
return params
|
