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