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
|
"""Handle linking of Phenotype data to the Auth(entic|oris)ation system."""
import uuid
from typing import Any, Iterable
from MySQLdb.cursors import DictCursor
import gn3.auth.db as authdb
import gn3.db_utils as gn3db
from gn3.auth.dictify import dictify
from gn3.auth.authorisation.checks import authorised_p
from gn3.auth.authorisation.groups.models import Group
def linked_phenotype_data(
authconn: authdb.DbConnection, gn3conn: gn3db.Connection,
species: str = "") -> Iterable[dict[str, Any]]:
"""Retrieve phenotype data linked to user groups."""
authkeys = ("SpeciesId", "InbredSetId", "PublishFreezeId", "PublishXRefId")
with (authdb.cursor(authconn) as authcursor,
gn3conn.cursor(DictCursor) as gn3cursor):
authcursor.execute("SELECT * FROM linked_phenotype_data")
linked = tuple(tuple(row[key] for key in authkeys)
for row in authcursor.fetchall())
if len(linked) <= 0:
return iter(())
paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(linked))
query = (
"SELECT spc.SpeciesId, spc.Name AS SpeciesName, iset.InbredSetId, "
"iset.InbredSetName, pf.Id AS PublishFreezeId, "
"pf.Name AS dataset_name, pf.FullName AS dataset_fullname, "
"pf.ShortName AS dataset_shortname, pxr.Id AS PublishXRefId "
"FROM "
"Species AS spc "
"INNER JOIN InbredSet AS iset "
"ON spc.SpeciesId=iset.SpeciesId "
"INNER JOIN PublishFreeze AS pf "
"ON iset.InbredSetId=pf.InbredSetId "
"INNER JOIN PublishXRef AS pxr "
"ON pf.InbredSetId=pxr.InbredSetId") + (
" WHERE" if (len(linked) > 0 or bool(species)) else "") + (
(" (spc.SpeciesId, iset.InbredSetId, pf.Id, pxr.Id) "
f"IN ({paramstr})") if len(linked) > 0 else "") + (
" AND"if len(linked) > 0 else "") + (
" spc.SpeciesName=%s" if bool(species) else "")
params = tuple(item for sublist in linked for item in sublist) + (
(species,) if bool(species) else tuple())
gn3cursor.execute(query, params)
return (item for item in gn3cursor.fetchall())
@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 ungrouped_phenotype_data(
authconn: authdb.DbConnection, gn3conn: gn3db.Connection):
"""Retrieve phenotype data that is not linked to any user group."""
with gn3conn.cursor() as cursor:
params = tuple(
(row["SpeciesId"], row["InbredSetId"], row["PublishFreezeId"],
row["PublishXRefId"])
for row in linked_phenotype_data(authconn, gn3conn))
paramstr = ", ".join(["(?, ?, ?, ?)"] * len(params))
query = (
"SELECT spc.SpeciesId, spc.SpeciesName, iset.InbredSetId, "
"iset.InbredSetName, pf.Id AS PublishFreezeId, "
"pf.Name AS dataset_name, pf.FullName AS dataset_fullname, "
"pf.ShortName AS dataset_shortname, pxr.Id AS PublishXRefId "
"FROM "
"Species AS spc "
"INNER JOIN InbredSet AS iset "
"ON spc.SpeciesId=iset.SpeciesId "
"INNER JOIN PublishFreeze AS pf "
"ON iset.InbredSetId=pf.InbredSetId "
"INNER JOIN PublishXRef AS pxr "
"ON pf.InbredSetId=pxr.InbredSetId")
if len(params) > 0:
query = query + (
f" WHERE (iset.InbredSetId, pf.Id, pxr.Id) NOT IN ({paramstr})")
cursor.execute(query, params)
return tuple(dict(row) for row in cursor.fetchall())
return tuple()
def __traits__(gn3conn: gn3db.Connection, params: tuple[dict, ...]) -> tuple[dict, ...]:
"""An internal utility function. Don't use outside of this module."""
if len(params) < 1:
return tuple()
paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(params))
with gn3conn.cursor(DictCursor) as cursor:
cursor.execute(
"SELECT spc.SpeciesId, iset.InbredSetId, pf.Id AS PublishFreezeId, "
"pf.Name AS dataset_name, pf.FullName AS dataset_fullname, "
"pf.ShortName AS dataset_shortname, pxr.Id AS PublishXRefId "
"FROM "
"Species AS spc "
"INNER JOIN InbredSet AS iset "
"ON spc.SpeciesId=iset.SpeciesId "
"INNER JOIN PublishFreeze AS pf "
"ON iset.InbredSetId=pf.InbredSetId "
"INNER JOIN PublishXRef AS pxr "
"ON pf.InbredSetId=pxr.InbredSetId "
"WHERE (spc.SpeciesName, iset.InbredSetName, pf.Name, pxr.Id) "
f"IN ({paramstr})",
tuple(
itm for sublist in (
(item["species"], item["group"], item["dataset"], item["name"])
for item in params)
for itm in sublist))
return cursor.fetchall()
@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_phenotype_data(
authconn:authdb.DbConnection, gn3conn: gn3db.Connection, group: Group,
traits: tuple[dict, ...]) -> dict:
"""Link phenotype traits to a user group."""
with authdb.cursor(authconn) as cursor:
params = tuple({
"data_link_id": str(uuid.uuid4()),
"group_id": str(group.group_id),
**item
} for item in __traits__(gn3conn, traits))
cursor.executemany(
"INSERT INTO linked_phenotype_data "
"VALUES ("
":data_link_id, :group_id, :SpeciesId, :InbredSetId, "
":PublishFreezeId, :dataset_name, :dataset_fullname, "
":dataset_shortname, :PublishXRefId"
")",
params)
return {
"description": (
f"Successfully linked {len(traits)} traits to group."),
"group": dictify(group),
"traits": params
}
|
