Data-Test-Executer Framework speziell zum Test von Datenverarbeitungen mit Datengenerierung, Systemvorbereitungen, Einspielungen, ganzheitlicher diversifizierender Vergleich
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

139 lines
5.3 KiB

#!/usr/bin/python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------------------------------------------
# Author : Ulrich Carmesin
# Source : gitea.ucarmesin.de
# ---------------------------------------------------------------------------------------------------------
import utils.config_tool
import tools.dbrel_tool
import mysql.connector
import basic.constants as B
import utils.data_const as D
import utils.date_tool
class DbFcts(tools.dbrel_tool.DbFcts):
"""
This interface defines each necessary connection to any kind of database.
The specific technique how to connect to the concrete DBMS has to be implemented in the specific tool.
"""
def __init__(self):
pass
def selectRows(self, table, job, where=""):
""" method to select rows from a database
statement written in sql """
tdata = {}
verify = -1+job.getDebugLevel("db_tool")
attr = self.getDbAttributes(B.SVAL_NULL)
sql = "SELECT * FROM "+attr[B.ATTR_DB_DATABASE]+"."+table
if len(where) > 3:
sql += " "+where
sql += ";"
self.comp.m.logInfo(sql)
connector = self.getConnector()
mycursor = connector.cursor()
mycursor.execute(sql)
myresult = mycursor.fetchall()
tdata[B.DATA_NODE_HEADER] = []
for f in self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_HEADER]:
tdata[B.DATA_NODE_HEADER].append(f)
tdata[B.DATA_NODE_DATA] = []
for x in myresult:
r = {}
i = 0
for f in self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_HEADER]:
if self.comp.conf[B.DATA_NODE_DDL][table][f][D.DDL_TYPE] in [D.TYPE_TIME, D.TYPE_DATE]:
r[f] = utils.date_tool.getFormatdate(x[i], utils.date_tool.F_DIR)
else:
r[f] = str(x[i])
i += 1
tdata[B.DATA_NODE_DATA].append(r)
self.comp.m.logInfo(str(tdata))
return tdata
def deleteRows(self, table, job, where=""):
""" method to delete rows from a database
statement written in sql """
verify = -1+job.getDebugLevel("db_tool")
attr = self.getDbAttributes(B.SVAL_NULL)
sql = "DELETE FROM "+attr[B.ATTR_DB_DATABASE]+"."+table
if len(where) > 3:
sql += " "+where
sql += ";"
self.comp.m.logInfo(sql)
self.execStatement(sql)
def updateRows(self, statement, job):
""" method to delete rows from a database
statement written in sql """
raise Exception(B.EXCEPT_NOT_IMPLEMENT)
def insertRows(self, table, rows, job):
""" method to insert rows into a database
the rows will be interpreted by the ddl of the component
"""
verify = -1+job.getDebugLevel("db_tool")
attr = self.getDbAttributes(B.SVAL_NULL)
insheader = self.getInsertFields(self.comp.conf[B.DATA_NODE_DDL][table])
if len(insheader) < len(self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_HEADER]):
lastid = 1
else:
lastid = 0
sql = "INSERT INTO "+attr[B.ATTR_DB_DATABASE]+"."+table
sql += " ( "+",".join(insheader) + " ) "
sql += " VALUES "
self.comp.m.logInfo(sql)
values = []
for r in rows:
rowvalues = []
for h in insheader:
if (h in r):
rowvalues.append("\'"+self.getDbValue(self.comp.conf[B.DATA_NODE_DDL][table][h], r[h])+"\'")
else:
rowvalues.append("\'"+self.getDbValue(self.comp.conf[B.DATA_NODE_DDL][table][h], "")+"\'")
sql += "("+",".join(rowvalues)+"), "
values.append(tuple(rowvalues))
sql = sql[0:-2]
self.comp.m.logInfo(str(values))
try:
connector = self.getConnector()
mycursor = connector.cursor()
mycursor.execute(sql)
if lastid > 0:
lastid = mycursor.lastrowid
connector.commit()
except Exception as e:
self.comp.m.setError("")
return 0
self.comp.m.setMsg(str(len(values))+" rows inserted into "+table)
return lastid
def execStatement(self, statement, conn=None):
""" add-on-method to execute the statement
this method should only called by the class itself """
connector = self.getConnector(conn)
cursor = connector.cursor()
try:
cursor.execute(statement)
connector.commit()
except Exception as e:
if "CREATE INDEX " in statement:
return
raise Exception("DB-Exception "+statement+"\n"+e.__str__())
print("Statement executed "+statement)
self.comp.m.setMsg("Statement executed")
def getConnector(self, conn=None):
""" add-on-method to get the connector
this method should only called by the class itself """
job = self.job # basic.program.Job.getInstance()
if conn is None:
conn = self.getDbAttributes(B.SVAL_NULL)
cnx = mysql.connector.connect(
host=conn[B.ATTR_DB_HOST],
user=conn[B.ATTR_DB_USER],
password=conn[B.ATTR_DB_PASSWD],
database=conn[B.ATTR_DB_DATABASE]
)
return cnx