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
|
"""Fetch data for datatables."""
import logging
from typing import Optional
from MySQLdb.cursors import DictCursor
from gn_libs.mysqldb import Connection, debug_query
logger = logging.getLogger(__name__)
def fetch_publications(
conn: Connection,
search: Optional[str] = None,
offset: int = 0,
limit: int = -1
) -> tuple[dict, int, int, int]:
"""Fetch publications from the database."""
_query = "SELECT * FROM Publication"
_count_query = "SELECT COUNT(*) FROM Publication"
_params = None
_where_clause = ""
_limit_clause = ""
if search is not None and bool(search):
_where_clause = ("WHERE PubMed_ID LIKE %s "
"OR Authors LIKE %s "
"OR Title LIKE %s")
_params = (f"%{search}%",) * 3
if limit > 0:
_limit_clause = f"LIMIT {limit} OFFSET {offset}"
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute("SELECT COUNT(*) FROM Publication")
_total_rows = int(cursor.fetchone()["COUNT(*)"])
cursor.execute(f"{_count_query} {_where_clause}", _params)
debug_query(cursor, logger)
_result = cursor.fetchone()
_total_filtered = int(_result["COUNT(*)"] if bool(_result) else 0)
cursor.execute(f"{_query} {_where_clause} {_limit_clause}", _params)
debug_query(cursor, logger)
_current_filtered = tuple(
{**dict(row), "index": idx}
for idx, row
in enumerate(cursor.fetchall(), start=offset+1))
return (
_current_filtered,
len(_current_filtered),
_total_filtered,
_total_rows)
|
