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
|
"""Phenotype data resources functions and utilities."""
import uuid
from typing import Optional, Sequence
import sqlite3
import gn_auth.auth.db.sqlite3 as db
from gn_auth.auth.authorisation.groups import Group
from .base import Resource
from .data import __attach_data__
def resource_data(
cursor: db.DbCursor,
resource_id: uuid.UUID,
offset: int = 0,
limit: Optional[int] = None) -> Sequence[sqlite3.Row]:
"""Fetch data linked to a Phenotype resource"""
cursor.execute(
("SELECT * FROM phenotype_resources AS pr "
"INNER JOIN linked_phenotype_data AS lpd "
"ON pr.data_link_id=lpd.data_link_id "
"WHERE pr.resource_id=?") + (
f" LIMIT {limit} OFFSET {offset}" if bool(limit) else ""),
(str(resource_id),))
return cursor.fetchall()
def link_data_to_resource(
conn: db.DbConnection,
resource: Resource,
group: Group,
data_link_id: uuid.UUID) -> dict:
"""Link Phenotype data with a resource."""
with db.cursor(conn) as cursor:
params = {
"group_id": str(group.group_id),
"resource_id": str(resource.resource_id),
"data_link_id": str(data_link_id)
}
cursor.execute(
"INSERT INTO phenotype_resources VALUES"
"(:group_id, :resource_id, :data_link_id)",
params)
return params
def unlink_data_from_resource(
conn: db.DbConnection,
resource: Resource,
data_link_id: uuid.UUID) -> dict:
"""Unlink data from Phenotype resources"""
with db.cursor(conn) as cursor:
cursor.execute("DELETE FROM phenotype_resources "
"WHERE resource_id=? AND data_link_id=?",
(str(resource.resource_id), str(data_link_id)))
return {
"resource_id": str(resource.resource_id),
"dataset_type": resource.resource_category.resource_category_key,
"data_link_id": str(data_link_id)
}
def attach_resources_data(
cursor, resources: Sequence[Resource]) -> Sequence[Resource]:
"""Attach linked data to Phenotype resources"""
placeholders = ", ".join(["?"] * len(resources))
cursor.execute(
"SELECT * FROM phenotype_resources AS pr "
"INNER JOIN linked_phenotype_data AS lpd "
"ON pr.data_link_id=lpd.data_link_id "
f"WHERE pr.resource_id IN ({placeholders})",
tuple(str(resource.resource_id) for resource in resources))
return __attach_data__(cursor.fetchall(), resources)
|
