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
|
"""Module to handle persistence and retrieval of publication to/from MariaDB"""
from gn_libs.mysqldb import Connection
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"
") "
"ON DUPLICATE KEY UPDATE "
"Abstract=VALUES(Abstract), Authors=VALUES(Authors), "
"Title=VALUES(Title), Journal=VALUES(Journal), "
"Volume=VALUES(Volume), Pages=VALUES(pages), "
"Month=VALUES(Month), Year=VALUES(Year) "
"RETURNING "),
publications)
paramstr = ", ".join(["%s"] * len(pubmed_ids))
cursor.execute(
("SELECT Id, PubMed_ID FROM Publication "
f"WHERE PubMed_ID IN ({paramstr})"),
pubmed_ids)
return {
row["PubMed_ID"]: row["Id"] for row in cursor.fetchall()
}
return {}
|
