121 lines
3.3 KiB
Python
121 lines
3.3 KiB
Python
import mariadb
|
|
import os
|
|
|
|
|
|
def getConnection():
|
|
try:
|
|
user = os.environ["DB_USER"]
|
|
password = os.environ["DB_PASS"]
|
|
host = os.environ["DB_HOST"]
|
|
database = os.environ["DB_NAME"]
|
|
|
|
conn = mariadb.connect(
|
|
user = user,
|
|
password = password,
|
|
host = host,
|
|
database = database
|
|
)
|
|
conn.autocommit = False
|
|
|
|
return conn
|
|
except mariadb.Error as err:
|
|
raise Exception("Error when connecting to database: {}".format(err))
|
|
except KeyError as err:
|
|
raise Exception("Database configuration variable {} not available".format(err))
|
|
|
|
|
|
|
|
def getMany(stmt, params, objName):
|
|
dbh = None
|
|
cur = None
|
|
try:
|
|
dbh = getConnection()
|
|
objs = []
|
|
cur = dbh.cursor(dictionary=True)
|
|
cur.execute(stmt, params)
|
|
for obj in cur:
|
|
objs.append(obj)
|
|
return objs
|
|
except Exception as err:
|
|
print("Error in getMany({}): {}".format(objName, err))
|
|
return str(err), 500
|
|
finally:
|
|
print("return connection in getMany({})".format(objName))
|
|
if cur:
|
|
cur.close()
|
|
if dbh:
|
|
dbh.close()
|
|
|
|
|
|
def getOne(stmt, params, objName):
|
|
dbh = None
|
|
cur = None
|
|
try:
|
|
dbh = getConnection()
|
|
cur = dbh.cursor(dictionary=True)
|
|
cur.execute(stmt, params)
|
|
obj = cur.next()
|
|
if not obj:
|
|
return "{} not found".format(objName), 404
|
|
invObj = cur.next()
|
|
if invObj:
|
|
return "More than one {} by that id ({}, {})".format(objName, id, invObj), 500
|
|
return obj
|
|
except Exception as err:
|
|
print("Error in getOne({}): {}".format(objName, err))
|
|
return str(err), 500
|
|
finally:
|
|
print("return connection in getOne({})".format(objName))
|
|
if cur:
|
|
cur.close()
|
|
if dbh:
|
|
dbh.close()
|
|
|
|
def putOne(stmt, params, objName):
|
|
dbh = None
|
|
cur = None
|
|
try:
|
|
dbh = getConnection()
|
|
cur = dbh.cursor(dictionary=True)
|
|
cur.execute(stmt, params)
|
|
dbh.commit()
|
|
return "{} successfully inserted with primary key {}".format(objName, cur.lastrowid), 202
|
|
except mariadb.Error as err:
|
|
dbh.rollback()
|
|
print("Database error in putOne({}): {}".format(objName, err))
|
|
return str(err), 500
|
|
except Exception as err:
|
|
dbh.rollback()
|
|
print("Error in putOne({}): {}".format(objName, err))
|
|
return str(err), 500
|
|
finally:
|
|
print("return connection in getOne({})".format(objName))
|
|
if cur:
|
|
cur.close()
|
|
if dbh:
|
|
dbh.close()
|
|
|
|
def call(procName):
|
|
dbh = None
|
|
cur = None
|
|
try:
|
|
dbh = getConnection()
|
|
cur = dbh.cursor(dictionary=True)
|
|
cur.execute("CALL {}(null)".format(procName))
|
|
dbh.commit()
|
|
return "{} successfully called".format(procName), 202
|
|
except mariadb.Error as err:
|
|
dbh.rollback()
|
|
print("Database error in call {}: {}".format(procName, err))
|
|
return str(err), 500
|
|
except Exception as err:
|
|
dbh.rollback()
|
|
print("Some error in call {}: {}".format(procName, err))
|
|
return str(err), 500
|
|
finally:
|
|
print("return connection in call {}".format(procName))
|
|
if cur:
|
|
cur.close()
|
|
if dbh:
|
|
dbh.close()
|