Datenbanken mit Python

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.

1Python Programmierkurs
2Python: Methoden
3Kontrollstrukturen
4Strings in Python
5Container
6Objekte in Python
7Module
8Exceptions in Python
9Typkonvertierung
10Python und Dateien
11Datum und Zeit mit Python verarbeiten
12Multithreading
13Netzwerk in Python
14Logging in Python
15GPIO
16Automatische Tests
17Datenbanken mit Python
18Python: Generatoren und List Comprehension
19Python: Webseiten mit Flask
20Python virtuelle Umgebungen
21Interrupts & Signale
22NumPy
23Matplotlib
24match

Hier geht es um die Grundlagen der Datenbankansteuerung in Python.

Python Logo (CC-BY-SA The people from the Tango! project / Wikipedia)
Datenbanken in Python
Python Logo (CC-BY-SA The people from the Tango! project / Wikipedia)

Vorbereitung

Ich benutze für diese Beispiele den DietPiOS Raspi auf dem ich den mariadb Client nachinstalliert habe.

sudo apt install mariadb-client

Danach logge ich mich auf dem MariaDB Server ein

mariadb -h terry -u root -p

und 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-python

Connector

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=pidb

Diese 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.

Schreibe einen Kommentar

Cookie Consent Banner von Real Cookie Banner