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
|
"""Module to handle persistence and retrieval of publication to/from MariaDB"""
import logging
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"
") "
"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)
return tuple({
**row, "PublicationId": 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()
|
