47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
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):
|
|
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)
|