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
|
"""Module to handle persistence and retrieval of publication to/from MariaDB"""
import logging
from typing import Iterable
from MySQLdb.cursors import DictCursor
from gn_libs.mysqldb import Connection, debug_query
logger = logging.getLogger(__name__)
def fetch_phenotype_publications(
conn: Connection,
ids: tuple[tuple[int, int], ...]
) -> tuple[dict, ...]:
"""Fetch publication from database by ID."""
paramstr = ",".join(["(%s, %s)"] * len(ids))
query = (
"SELECT "
"pxr.PhenotypeId, pxr.Id AS xref_id, pxr.PublicationId, pub.PubMed_ID "
"FROM PublishXRef AS pxr INNER JOIN Publication AS pub "
"ON pxr.PublicationId=pub.Id "
f"WHERE (pxr.PhenotypeId, pxr.Id) IN ({paramstr})")
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute(query, tuple(item for row in ids for item in row))
return tuple(dict(row) for row in cursor.fetchall())
def create_new_publications(
conn: Connection,
publications: tuple[dict, ...]
) -> tuple[dict, ...]:
if len(publications) > 0:
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.executemany(
("INSERT INTO "
"Publication( "
"PubMed_ID, Abstract, Authors, Title, Journal, Volume, Pages, "
"Month, Year"
") "
"VALUES("
"%(pubmed_id)s, %(abstract)s, %(authors)s, %(title)s, "
"%(journal)s, %(volume)s, %(pages)s, %(month)s, %(year)s"
") "
"RETURNING *"),
publications)
return tuple({
**row, "publication_id": row["Id"]
} for row in cursor.fetchall())
return tuple()
def update_publications(conn: Connection , publications: tuple[dict, ...]) -> tuple[dict, ...]:
"""Update details for multiple publications"""
if len(publications) > 0:
with conn.cursor(cursorclass=DictCursor) as cursor:
logger.debug("UPDATING PUBLICATIONS: %s", publications)
cursor.executemany(
("UPDATE Publication SET "
"PubMed_ID=%(pubmed_id)s, Abstract=%(abstract)s, "
"Authors=%(authors)s, Title=%(title)s, Journal=%(journal)s, "
"Volume=%(volume)s, Pages=%(pages)s, Month=%(month)s, "
"Year=%(year)s "
"WHERE Id=%(publication_id)s"),
publications)
debug_query(cursor, logger)
return publications
return tuple()
return tuple()
def fetch_publications(conn: Connection) -> Iterable[dict]:
"""Fetch publications from the database."""
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute("SELECT * FROM Publication")
for row in cursor.fetchall():
yield dict(row)
def fetch_publication_by_id(conn: Connection, publication_id: int) -> dict:
"""Fetch a specific publication from the database."""
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute("SELECT * FROM Publication WHERE Id=%s",
(publication_id,))
return dict(cursor.fetchone())
def fetch_publication_phenotypes(
conn: Connection, publication_id: int) -> Iterable[dict]:
"""Fetch all phenotypes linked to this publication."""
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute(
"SELECT pxr.Id AS xref_id, pxr.PublicationId, phe.* "
"FROM PublishXRef AS pxr INNER JOIN Phenotype AS phe "
"ON pxr.PhenotypeId=phe.Id "
"WHERE pxr.PublicationId=%s",
(publication_id,))
while True:
row = cursor.fetchone()
if row:
yield row
else:
break
|
