Datenbanken müssen dir keine Ehrfurcht einflössen. Ich habe die bereits gezeigt, wie du MariaDB in einem Docker Container installierst, jetzt geht es darum, diese Datenbank mit Python zu benutzen. Ich werde in diesem Blog-Post kein SQL-Tutorial schreiben, dazu gibt es genügend im Web zu finden.
Hier geht es um die Grundlagen der Datenbankansteuerung in Python.
Vorbereitung
Ich benutze für diese Beispiele den DietPiOS Raspi auf dem ich den mariadb Client nachinstalliert habe.
sudo apt install mariadb-clientDanach logge ich mich auf dem MariaDB Server ein
mariadb -h terry -u root -pund lege Benutzer und Datenbank sowie die Zugriffsrechte an.
create user 'piuser'@'192.168.178.%' identified by '<passwort>';
create database pidb;
grant all privileges on pidb.* to 'piuser'@'192.168.178.%';
flush privileges;Um mit Python die DB anzusteuern benötigen wir das mysql Connector Modul auf dem Raspberry Pi.
pip3 install mysql-connector-pythonConnector
Die Klasse DBConnector verwaltet für uns die Verbindung zur Datenbank und stellt die notwendigen Methoden zur Verfügung.
#encoding: utf-8
import mysql.connector
import logging
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
class DBConnection():
def __init__(self):
self.dbconnection = None
#Verbindung zur Datenbank mit den übergebenen Wreten herstellen.
def connect(self, host, user, password, database):
self.connection= mysql.connector.connect(host=host, user=user,password=password,database=database)
# Verbindung trennen
def disconnect(self):
if self.connection is not None:
self.connection.close()
self.connection = None
# Schüler-Record in die Tabelle einfügen.
def insertSchueler(self,schueler):
self.execute_query('insert into schueler (name,vorname) values(%s,%s)', (schueler['name'],schueler['vorname']))
# Liefert True, wenn die Verbindung zur DB besteht, sonst False
def isConnected(self):
if self.connection is not None:
return self.connection.is_connected()
return False
# Wenn DB-Verbindung besteht, wird der Cursor geliefert, sonst None
def cursor(self):
if self.isConnected():
return self.connection.cursor()
return None
# Datenbank Kommando ausführen. Das params-Tupel wird in die Query eingesetzt.
# Liefert ein Resultset zurück.
def execute_query(self, query, params = None):
if self.isConnected():
cursor = self.cursor()
if cursor is not None:
cursor.execute(query,params) #Kommdo ausführen
return cursor.fetchall() # Ergebnis zurückliefern.
In der Praxis ist es üblich, die Verbindung zur Datenbank immer offen zu halten. Meistens macht dies eine Middleware, die durch unseren DBConnector simuliert wird. Das Aufbau der Verbindung ist zeitlich vergleichsweise teuer, daher empfiehlt es sich, diese erst zu trennen, wenn die Datenbank nicht mehr benötigt wird. Dazu später mehr.
Datenbanken programmieren.
Um die Möglichkeiten des Datenbankzugriffs zu verdeutlichen, habe ich mich entschlossen, einen Unit-Test dazu zu schreiben.
Wenn du jetzt denkst, die Verbindung zur Datenbank bauen wir in der setUp() Methode auf, muss ich dich enttäuschen. Da setUp() vor jeder Testmethode aufgerufen, d.h. du hast nachher eine Menge ungenutzter Verbindungen zur Datenbank rumliegen, die deinen Datenbankserver irgendwann überlasten. Ich benutze hier die setUpClass() Methode, die als Klassenmethode nur beim Laden der Klasse aufgerufen wird.
@classmethod
def setUpClass(cls):
logging.debug(f'setUpClass() {cls}')
config = configparser.ConfigParser()
config.read('db.ini')
logging.debug(f"host={config['pidb']['host']}")
cls.db_connection = DBConnection()
cls.db_connection.connect(config['pidb']['host'],config['pidb']['user'],config['pidb']['password'],config['pidb']['database'])
if cls.db_connection.isConnected():
logging.debug('erzeuge Table schueler...')
cls.db_connection.execute_query('create table if not exists schueler (id int auto_increment primary key, vorname varchar(20),name varchar(20))')
# Testdaten aufbauen.
cls.fill_table()
logging.debug('setup() fertig')
Ich sehe häufig Beispielprogramme, in denen die Loginparameter zur DB im Code fest verankert sind. Dies ist nicht nur unschön, sondern auch ein Sicherheitsrisiko. Daher legen wir eine Datei db.ini an.
[pidb]
host=database
user=piuser
password=<passwort>
database=pidbDiese Datei ist eine Registry von Key/Value Paaren, die mit dem Modul configparser, sehr einfach eingelesen werden kann.
Der Decorator @classmethod kennzeichnet unsere setUpClass() Methode als Klassenmethode. In der Methode wird die db.ini Datei eingelesen und mit den Werten darin die DBConnection hergestellt. Danach wird die Tabelle schueler als Testtabelle angelegt, sofern sie noch nicht existiert. Dann wird die Tabelle mit Testdaten bestückt. Dazu dient die Methode fill_table(), die eine Liste von Dictionaries durchläuft und einen Datensatz pro Listenelement anlegt.
schueler = [
{'name':'Simpson','vorname': 'Bart'},
{'name':'Simpson','vorname':'Lisa'},
{'name':'van Houten','vorname':'Milhouse'},
{'name':'Wiggum','vorname':'Ralph'},
{'name':'Jones','vorname':'Jimbo'}
]
#Testdaten in die Datenbank übertragen
@classmethod
def fill_table(cls):
for s in range(0,len(schueler)):
student = schueler[s]
cls.db_connection.insertSchueler(student)
Da fill_table() von einer Klassenmethode aufgerufen wird, muss auch selber als Klassenmethode dekoriert werden. Beachte, dass Klassenmethoden keinen self Parameter übergeben bekommen, sondern einen cls Parameter. Dies liegt daran, dass es kein Objekt gibt, dass self referenzieren könnte sondern nur der Klassenkontext.
Jetzt testen wir, ob alle Datensätze angelegt wurden, in dem wir alle Einträge der Tabelle Schüler lesen.
def test_select_all(self):
# self.test_insert()
result = self.db_connection.execute_query("select * from schueler")
logging.debug(f'result={result}')
self.assertEqual(len(schueler),len(result))
for i in range(len(result)):
self.assertEqual(schueler[i]['vorname'], result[i][1])
Hier wird nicht nur getestet, ob die Anzahl der gelieferten Records mit der Anzahl der Elemente der schueler-Liste übereinstimmt, sondern auch, dass alle Vornamen vorhanden sind.
In einer weiteren Testmethode lassen wir uns nur die Records liefern, die den Nachnamen „Simpson“ haben und prüfen, ob es genau zwei Einträge sind.
# Nur die Records mit Nachname ='Simpson' lesen
def test_select_name(self):
self.assertTrue(self.db_connection.isConnected())
result = self.db_connection.execute_query("select * from schueler where name='Simpson'")
self.assertEqual(2,len(result))
Test aufräumen
Nach Abschluss der Tests müssen wir noch aufräumen. Dazu benutzen wir hier die Klassenmethode tearDownClass(), die analog zu setUpClass() funktioniert.
@classmethod
def tearDownClass(cls):
logging.debug(f'tearDownClass() {cls}')
logging.debug('Räume Tables auf')
cls.db_connection.execute_query('drop table if exists schueler')
logging.debug('schliesse Datenbankverbindung.')
cls.db_connection.disconnect()
cls.db_connection = None
Wenn die Tabelle schueler existiert, wird sie verworfen, dann wird die Verbindung zur Datenbank geschlossen und das Connector Objekt verworfen.
Fazit Datenbanken
Datenbanken speichern sehr effizient und können relativ einfach in Python benutzt werden.
Ich hatte den Test oben zunächst mit setUp() und tearDown() realisiert. Nach der Umstellung auf setUpClass() und tearDownClass() lief der Test fast 4/10 Sekunden schneller. Das klingt nicht nach viel, aber stell dir mal ein ein Programm vor, dass mehrere 1000 Queries an die Datenbank schickt.
Im nächsten Teil werden wir uns dann mit Generatoren beschäftigen.
