"""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[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)