hv2-all-in-one/cli/db.py

import psycopg2
import psycopg2.extras
from loguru import logger


class NoDataFoundException(Exception): pass

class TooMuchDataFoundException(Exception): pass


def execDatabaseOperation(dbh, func, params):
    cur = None
    try:
        with dbh.cursor(cursor_factory = psycopg2.extras.RealDictCursor) as cur:
            # params["params"] = [ v if not v=='' else None for v in params["params"] ]
            logger.debug("edo: {}".format(str(params)))
            return func(cur, params)
    except psycopg2.Error as err:
        raise Exception("Error when working on cursor: {}".format(err))



def _opGetMany(cursor, params):
    #logger.warning(f"{params=}")
    items = []
    cursor.execute(params["statement"], params["params"])
    for itemObj in cursor:
        logger.debug("item received {}".format(str(itemObj)))
        items.append(itemObj)
    return items

def dbGetMany(dbh, params):
    return execDatabaseOperation(dbh, _opGetMany, params)

def _opGetOne(cursor, params):
    cursor.execute(params["statement"], params["params"])
    itemObj = cursor.fetchone()
    logger.debug(f"item received: {itemObj}")
    if not itemObj:
        raise NoDataFoundException
    dummyObj = cursor.fetchone()
    if dummyObj:
        raise TooMuchDataFoundException
    return itemObj

def dbGetOne(dbh, params):
    return execDatabaseOperation(dbh, _opGetOne, params)