Python Datenbank Klasse

Für Python Datenbank Projekte habe ich eine Klasse geschrieben, die sich um die Verbindung und (bisher) Select und Insert Abfragen kümmert. Beim Definieren wird die Konstruktormethode aufgerufen und die Verbindungsparameter übergeben. Die Objektvariablen bekommen die jeweilige Eingaben zugewiesen.

def __init__(self,host,username,password,database): 
    self.host = host 
    self.username = username 
    self.password = password 
    self.database = database

Connect Methode

Die connect Methode benutzt die Objektvariablen und baut mit diesen Informationen die Verbindung zu der Datenbank auf. self.cnx speichert die erzeugte Session. Somit steht diese Session dem Code weiter zur Verfügung und kann vom nachfolgenden Code mit der cnx Variable aufgerufen werden.  Im Falle von Fehlern bei der Verbindung springt der Code in den Exception Block.

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"

Query_Select Methode

In der query_select Methode werden, wie der Name schon sagt, die Select Abfragen ausgeführt. Die Funktion erwartet das SQL-Statement und falls vorhanden, die Parameters, die z.B. in der where clause mit angegeben werden können. Die IF-Anweisung überprüft dementsprechend, ob Parameters vorhanden sind und springt zum gültigen Programmblock. Bei der cursor.execute() Methode wird das Statement mit dem übergegebenen Parametern ausgeführt, während cursor.fetchall() alle Ergebnisse zurückliefert und in rtnvalue abspeichert. Tritt eine Exception auf, wird eine Fehlermeldung ausgegeben und die Funktion beendet sich mit einem False State.

def query_select(self,sqlstatement, params): 
    if (params is not None): 
        try: 
            self.cursor.execute(sqlstatement, (params,)) 
            rtnvalue = self.cursor.fetchall() 
            print colored("[I] + executed select with params","green") 
            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","green") 
            return rtnvalue 
        except: 
            print colored("[C] + something went wrong while executing select without params","red") 
            return None

Query_Insert Methode

Ähnlich wie die query_select Funktion funktioniert die query_Insert. Die Methode erwartet die Parameter als Dictionary, andernfalls kann die Funktion nicht richtig abgearbeitet werden, da zwei Variablen mit der key() beziehungsweise values() methode gefüttert werden. Diese genannten Methoden stehen nur Dictionaries zur Verfügung. In der sql Variable wird die Insert Query noch mit dem Spaltennamen in “column” und die dazügehörigen Werte in “join(repr(e) for e in Values)” gefüttert. cnx.commit  überträgt die Insert Einträge in die Datenbank. Falls das commit abbricht, springt der Code in die Exception und cnx.rollback() rollt den Insert wieder zurück. Abschließend beendet sich die Funktion mit einem False Statement.

def query_insert(self, sqlstatement, params):
    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") 
        self.cursor.execute(sql) 
    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: " + err.erno,"red") 
        return False

Disconnect Methode

Zum Schluss schliesst der Code die aktuelle Python Datenbankverbindung mit cnx.close().

def disconnect(self): 
   self.cnx.close() 
   print colored("[I] + Connection closed","green")

Vollständiger Code

Hier folgt der Code ohne Unterbrechung und Beschreibung:

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 query_select(self,sqlstatement, params):
         if (params is not None):
             try:
                 self.cursor.execute(sqlstatement, (params,))
                 rtnvalue = self.cursor.fetchall()
                print colored("[I] + executed select with params","green")
                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","green")
                return rtnvalue
            except:
                print colored("[C] + something went wrong while executing select without params","red")
                return None

    def query_insert(self, sqlstatement, params):
        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")
            self.cursor.execute(sql)
        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: " + err.erno,"red")
            return False

    def disconnect(self):
        self.cnx.close()
        print colored("[I] + Connection closed","green")