48 lines
1.3 KiB
Python
Raw Normal View History

2021-09-13 19:27:32 +02:00
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:
2021-12-19 14:17:44 +01:00
# params["params"] = [ v if not v=='' else None for v in params["params"] ]
2021-09-13 19:27:32 +02:00
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):
2021-12-19 14:17:44 +01:00
logger.warning(f"{params=}")
2021-09-13 19:27:32 +02:00
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)