steviesblog.de

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"))