In meiner Datenbank Klasse habe ich lange Zeit die Update Funktion missen lassen. Hier gelangt ihr zum alten Artikel. Einfach aus dem Grund weil ich sie vorher noch nicht benötigt habe. Das habe ich hiermit nun nachgezogen(Eigentlich existiert sie schon wesentlich länger, aber ich habe jetzt erst den Artikel verfasst 😉 ). Das Update kann separat aufgerufen werden, macht aber auch Sinn in Zusammenhang mit einem Insert. Falls dort z.B. ein Primärschlüssel schon existiert, kann dieser geupdated werden.
Für ein internes Projekt habe ich folgende Befehlszeilen dafür benutzt
if not DB.QueryInsert("insert into AUX_PROJECTS(%s) VALUES(%s)",{"ID":PList[i][0],PList[0][1]:PList[i][1],"ZINS":str(PList[i][2]).replace(",","."),PList[0][3]:PList[i][3],"ANLAGE":PList[i][4],"STATUS":PList[i][5],"SCORE":PList[i][6],PList[0][7]:PList[i][7],PList[0][8]:PList[i][8],"UNIT":PList[i][9],"UPLOADED_AT":PList[i][10]}, True):
DB.QueryUpdate("update AUX_PROJECTS SET %s where %s",{PList[0][1]:PList[i][1],"ZINS":str(PList[i][2]).replace(",","."),"ANLAGE":PList[i][4]},"ID", PList[i][0])
Das Insert erwahrtet wie gewohnt die insert Query gefolgt von einem Tupel mit den Namen und den Werten. Als 3.ten Parameter kann noch das Update auf True oder False gesetzt werden, dies triggert allerdings zurzeit nur eine Ausgabe. Wenn dieses Insert nicht funktioniert, bzw. ein False zurückgibt, versucht er ein Update. Der Aufbau ist ähnlich wie beim Insert. Zuerst die Update Query, gefolgt von dem Tupel.
Eine Update könnte dann wie folgt in der Shell angezeigt werden:
Hier der komplette neue Code.
import mysql.connector
from mysql.connector import errorcode
from termcolor import colored
class databaseconnection:
host=username=password=database=cnx=cursor=None
def __init__(self,host,username,password,database):
self.host = host
self.username = username
self.password = password
self.database = database
def Connect(self):
config = {
'user': self.username,
'password': self.password,
'host': self.host,
'database': self.database
}
try:
self.cnx = mysql.connector.connect(**config)
self.cursor = self.cnx.cursor()
print(colored("[I] + DB Connection established","green"))
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print(colored("[C] + Wrong Username/Password","red"))
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print(colored("[C] + Database Error","red"))
else:
print(colored("[C] + " + err,"red"))
def QuerySelect(self,sqlstatement, params, header):
if (params is not None):
try:
self.cursor.execute(sqlstatement, (params,))
rtnvalue = self.cursor.fetchall()
print(colored("[I] + executed select with params","green"))
if header:
rtnvalue.insert(0,self.cursor.column_names)
return rtnvalue
except:
print(colored("[C] + something went wrong while executing select with params:","red"))
return False
else:
try:
self.cursor.execute(sqlstatement)
rtnvalue = self.cursor.fetchall()
print(colored("[I] + executed select without params: " + sqlstatement,"green"))
if header:
rtnvalue.insert(0,self.cursor.column_names)
return rtnvalue
except:
print(colored("[C] + something went wrong while executing select without params","red"))
return None
def QueryInsert(self, sqlstatement, params, update):
if (params is not None):
columns = ', '.join(params.keys())
values = params.values()
sql = sqlstatement % (columns, ', '.join(repr(e) for e in values))
print(colored("[I] + " + sql,"yellow"))
try:
self.cursor.execute(sql)
except mysql.connector.Error as er:
if er.errno == errorcode.ER_DUP_ENTRY:
if(update == True):
print(colored("[W] + transaction rolled back: Duplication found - Try update " ,"yellow"))
else:
print(colored("[W] + transaction rolled back: Duplication found " ,"yellow"))
###########try to make an update##########
return False
else:
print(colored("[C] + a criticial error occured " + er.msg,"red"))
return False
else:
print(colored("[C] + insert execution failed due missing parameters","red"))
return False
try:
self.cnx.commit()
print(colored("[I] + transaction committed, value added","green"))
return True
except mysql.connector.Error as er:
self.cnx.rollback()
print(colored("[C] + transaction rolled back: " + er.msg,"red"))
return False
# performs an DB update
# @var sqlstatement (update AUX_PROJECTS SET %s where %s)
# @var params Data to being updated
# @var Conditionheader Primary key
# @var ConditionValue PK agains value
def QueryUpdate(self, sqlstatement, params, Conditionheader,ConditionValue):
if (params is not None):
updatestring = self.CreateUpdateString(params)
sql = sqlstatement % (updatestring,Conditionheader + " = " + ConditionValue)
print(colored("[I] + " + sql,"yellow"))
try:
self.cursor.execute(sql)
except mysql.connector.Error as er:
self.cnx.rollback()
print(colored("[C] + transaction rolled back: " + er.msg,"red"))
return False
else:
print(colored("[C] + update execution failed due missing parameters","red"))
return False
try:
self.cnx.commit()
print(colored("[I] + transaction committed, value updated","green"))
return True
except mysql.connector.Error as er:
self.cnx.rollback()
print(colored("[C] + transaction rolled back: " + er.msg,"red"))
return False
def CreateUpdateString(self,params):
updatestring = ''
for key,value in params.items():
updatestring = updatestring + key + '=' + repr(str(value))
if(key == list(params.keys())[-1]):
continue
updatestring = updatestring + ','
return str(updatestring)
def Disconnect(self):
self.cnx.close()
print(colored("[I] + Connection closed","green"))