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 mRNA Assay data to the Auth(entic|oris)ation system."""
import uuid
import logging
from typing import Iterable
from functools import reduce
from dataclasses import asdict
from gn_libs import mysqldb as gn3db
from MySQLdb.cursors import DictCursor
from gn_auth.auth.db import sqlite3 as authdb
from gn_auth.auth.authorisation.checks import authorised_p
from gn_auth.auth.authorisation.resources.groups.models import Group
logger = logging.getLogger(__name__)
def linked_mrna_data(conn: authdb.DbConnection) -> Iterable[dict]:
"""Retrieve mRNA Assay data that is linked to user groups."""
with authdb.cursor(conn) as cursor:
cursor.execute("SELECT * FROM linked_mrna_data")
return (dict(row) for row in 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 ungrouped_mrna_data(# pylint: disable=[too-many-arguments, too-many-positional-arguments]
authconn: authdb.DbConnection, gn3conn: gn3db.Connection,
search_query: str, selected: tuple[dict, ...] = tuple(),
limit: int = 10000, offset: int = 0) -> tuple[
dict, ...]:
"""Retrieve mrna data that is not linked to any user group."""
params = tuple(
(row["SpeciesId"], row["InbredSetId"], row["ProbeFreezeId"],
row["ProbeSetFreezeId"])
for row in linked_mrna_data(authconn)) + tuple(
(row["SpeciesId"], row["InbredSetId"], row["ProbeFreezeId"],
row["ProbeSetFreezeId"])
for row in selected)
query = (
"SELECT s.SpeciesId, iset.InbredSetId, iset.InbredSetName, "
"pf.ProbeFreezeId, pf.Name AS StudyName, psf.Id AS ProbeSetFreezeId, "
"psf.Name AS dataset_name, psf.FullName AS dataset_fullname, "
"psf.ShortName AS dataset_shortname "
"FROM Species AS s INNER JOIN InbredSet AS iset "
"ON s.SpeciesId=iset.SpeciesId INNER JOIN ProbeFreeze AS pf "
"ON iset.InbredSetId=pf.InbredSetId INNER JOIN ProbeSetFreeze AS psf "
"ON pf.ProbeFreezeId=psf.ProbeFreezeId ") + (
"WHERE " if (len(params) > 0 or bool(search_query)) else "")
if len(params) > 0:
paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(params))
query = query + (
"(s.SpeciesId, iset.InbredSetId, pf.ProbeFreezeId, psf.Id) "
f"NOT IN ({paramstr}) "
) + ("AND " if bool(search_query) else "")
if bool(search_query):
query = query + (
"CONCAT(pf.Name, psf.Name, ' ', psf.FullName, ' ', psf.ShortName) "
"LIKE %s ")
params = params + ((f"%{search_query}%",),)# type: ignore[operator]
query = query + f"LIMIT {int(limit)} OFFSET {int(offset)}"
with gn3conn.cursor(DictCursor) as cursor:
cursor.execute(
query, tuple(item for sublist in params for item in sublist))
return tuple(row for row in 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_mrna_data(
conn: authdb.DbConnection, group: Group, datasets: dict) -> dict:
"""Link genotye `datasets` to `group`."""
with authdb.cursor(conn) as cursor:
cursor.executemany(
"INSERT INTO linked_mrna_data VALUES "
"(:data_link_id, :group_id, :SpeciesId, :InbredSetId, "
":ProbeFreezeId, :ProbeSetFreezeId, :dataset_name, "
":dataset_fullname, :dataset_shortname) "
"ON CONFLICT "
"(SpeciesId, InbredSetId, ProbeFreezeId, ProbeSetFreezeId) "
"DO NOTHING",
tuple({
"data_link_id": str(uuid.uuid4()),
"group_id": str(group.group_id),
**{
key: value for key,value in dataset.items() if key in (
"SpeciesId", "InbredSetId", "ProbeFreezeId",
"ProbeSetFreezeId", "dataset_fullname", "dataset_name",
"dataset_shortname")
}
} for dataset in datasets))
return {
"description": (
f"Successfully linked {len(datasets)} to group "
f"'{group.group_name}'."),
"group": asdict(group),
"datasets": datasets
}
def resources_by_datasets_and_traits(
authconn: authdb.DbConnection,
dsets_traits: tuple[tuple[str, str], ...]
) -> tuple[dict, ...]:
"""Fetch resources by their attached datasets and traits."""
traits_by_datasets: dict[str, tuple[str, ...]] = reduce(
lambda acc, curr: {
**acc,
curr[0]: acc.get(curr[0], tuple()) + (curr[1],)
},
dsets_traits,
{})
paramstr = ", ".join(["?"] * len(dsets_traits))
query = (
"SELECT r.*, rc.*, lmd.dataset_name FROM linked_mrna_data AS lmd "
"INNER JOIN mrna_resources AS mr ON lmd.data_link_id=mr.data_link_id "
"INNER JOIN resources AS r ON mr.resource_id=r.resource_id "
"INNER JOIN resource_categories AS rc "
"ON r.resource_category_id=rc.resource_category_id "
"WHERE lmd.dataset_name "
f"IN ({paramstr})")
logger.debug("QUERY: %s", query)
with authdb.cursor(authconn) as cursor:
cursor.execute(query, tuple(traits_by_datasets.keys()))
return tuple({
"resource_id": row["resource_id"],
"resource_data": tuple(
f'{row["dataset_name"]}::{trait_id}'
for trait_id in traits_by_datasets[row["dataset_name"]])
} for row in cursor.fetchall())
|
