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
|
"""Functions and utilities to handle ProbeSets from the database."""
from typing import Optional
from dataclasses import dataclass
from MySQLdb.cursors import DictCursor
from gn3.db_utils import Connection as DBConnection
from .query_tools import mapping_to_query_columns
@dataclass(frozen=True)
class Probeset: # pylint: disable=[too-many-instance-attributes]
"""Data Type that represents a Probeset"""
id_: Optional[int] = None
name: Optional[str] = None
symbol: Optional[str] = None
description: Optional[str] = None
probe_target_description: Optional[str] = None
chr_: Optional[str] = None
mb: Optional[float] = None # pylint: disable=C0103
alias: Optional[str] = None
geneid: Optional[str] = None
homologeneid: Optional[str] = None
unigeneid: Optional[str] = None
omim: Optional[str] = None
refseq_transcriptid: Optional[str] = None
blatseq: Optional[str] = None
targetseq: Optional[str] = None
strand_probe: Optional[str] = None
probe_set_target_region: Optional[str] = None
probe_set_specificity: Optional[float] = None
probe_set_blat_score: Optional[float] = None
probe_set_blat_mb_start: Optional[float] = None
probe_set_blat_mb_end: Optional[float] = None
probe_set_strand: Optional[str] = None
probe_set_note_by_rw: Optional[str] = None
flag: Optional[str] = None
# Mapping from the Phenotype dataclass to the actual column names in the
# database
probeset_mapping = {
"id_": "Id",
"name": "Name",
"symbol": "symbol",
"description": "description",
"probe_target_description": "Probe_Target_Description",
"chr_": "Chr",
"mb": "Mb",
"alias": "alias",
"geneid": "GeneId",
"homologeneid": "HomoloGeneID",
"unigeneid": "UniGeneId",
"omim": "OMIM",
"refseq_transcriptid": "RefSeq_TranscriptId",
"blatseq": "BlatSeq",
"targetseq": "TargetSeq",
"strand_probe": "Strand_Probe",
"probe_set_target_region": "Probe_set_target_region",
"probe_set_specificity": "Probe_set_specificity",
"probe_set_blat_score": "Probe_set_BLAT_score",
"probe_set_blat_mb_start": "Probe_set_Blat_Mb_start",
"probe_set_blat_mb_end": "Probe_set_Blat_Mb_end",
"probe_set_strand": "Probe_set_strand",
"probe_set_note_by_rw": "Probe_set_Note_by_RW",
"flag": "flag"
}
def fetch_probeset_metadata_by_name(conn: DBConnection, name: str) -> dict:
"""Fetch a ProbeSet's metadata by its `name`."""
with conn.cursor(cursorclass=DictCursor) as cursor:
cols = ", ".join(mapping_to_query_columns(probeset_mapping))
cursor.execute((f"SELECT {cols} "
"FROM ProbeSet "
"WHERE Name = %(name)s"),
{"name": name})
return cursor.fetchone()
def update_probeset(conn, probeset_id, data:dict) -> int:
"""Update the ProbeSet table with given `data`."""
cols = ", ".join(f"{probeset_mapping[col]}=%({col})s"
for col in data
if (col not in ("id_", "id") and
col in probeset_mapping))
if not bool(cols):
return 0
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute(
f"UPDATE ProbeSet SET {cols} WHERE Id=%(probeset_id)s",
{"probeset_id": probeset_id, **data})
return cursor.rowcount
|
