Browse Source

Teil17 Datenbanken

develop teil17
Olli Graf 1 year ago
parent
commit
9e78b44184
  1. 65
      db/testdb.py
  2. 1
      teil17/.gitignore
  3. 34
      teil17/DBConnection.py
  4. BIN
      teil17/__pycache__/DBConnection.cpython-39.pyc
  5. 0
      teil17/connectdb.py
  6. 8
      teil17/db.ini.sample
  7. 88
      teil17/testdb.py

65
db/testdb.py

@ -1,65 +0,0 @@
import unittest
from DBConnection import DBConnection
import logging
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
class TestDBConnection(unittest.TestCase):
@classmethod
def setUpClass(cls):
logging.debug(f'setUpClass() {cls}')
cls.db_connection = DBConnection()
cls.db_connection.connect('database','piuser','SW_Ep.6:ROTJ','pidb')
logging.debug('erzeuge Table fib...')
cls.db_connection.execute_query('create table if not exists fib (id int auto_increment primary key, wert int)')
cls.db_connection.execute_query('create table if not exists schueler (id int auto_increment primary key, vorname varchar(20),name varchar(20))')
logging.debug('setup() fertig')
@classmethod
def tearDownClass(cls):
logging.debug(f'tearDownClass() {cls}')
logging.debug('Räume Tables auf')
cls.db_connection.execute_query('drop table if exists fib')
cls.db_connection.execute_query('drop table if exists schueler')
logging.debug('schliesse Datenbankverbindung.')
cls.db_connection.disconnect()
cls.db_connection = None
def test_insert(self):
logging.debug(f'connected= {self.db_connection.isConnected()}')
self.assertTrue(self.db_connection.isConnected())
self.db_connection.execute_query("insert into schueler (name,vorname) values('Simpson','Bart')")
self.db_connection.execute_query("insert into schueler (name,vorname) values('Simpson','Lisa')")
self.db_connection.execute_query("insert into schueler (name,vorname) values('van Houten','Milhouse')")
self.db_connection.execute_query("insert into schueler (name,vorname) values('Wiggum','Ralph')")
# input('Press Any Key')
def test_select(self):
# self.test_insert()
self.assertTrue(self.db_connection.isConnected())
result = self.db_connection.execute_query("select * from schueler where name='Simpson'")
self.assertEqual(2,len(result))
result = self.db_connection.execute_query("select * from schueler")
self.assertEqual(4,len(result))
def test_connected(self):
self.assertTrue(self.db_connection.isConnected())
def test_disconnect(self):
self.db_connection.disconnect()
self.assertFalse(self.db_connection.isConnected())
def test_create_table(self):
self.assertTrue(self.db_connection.isConnected())
if __name__ =='__main__':
unittest.main()

1
db/.gitignore → teil17/.gitignore

@ -1,2 +1,3 @@
./__pycache__
./db.ini
./network/__pycache__/*

34
db/DBConnection.py → teil17/DBConnection.py

@ -1,3 +1,4 @@
#encoding: utf-8
import mysql.connector
import logging
@ -7,47 +8,38 @@ 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)
logging.debug(f'Datenbankverbindung aufgebaut ${self.connection}')
logging.debug(f'Cursor gesetzt ${self.cursor}')
# 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):
# logging.debug(f'connection={self.connection}')
# logging.debug(f'connected= {self.connection.is_connected()}')
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
def insert(self,cmd):
pass
def fetch(self):
pass
def create_table(self):
pass
# 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)
return cursor.fetchall()
cursor.execute(query,params) #Kommdo ausführen
return cursor.fetchall() # Ergebnis zurückliefern.

BIN
teil17/__pycache__/DBConnection.cpython-39.pyc

Binary file not shown.

0
db/connectdb.py → teil17/connectdb.py

8
teil17/db.ini.sample

@ -0,0 +1,8 @@
[pidb]
host=database
user=piuser
password=<passwort>
database=pidb

88
teil17/testdb.py

@ -0,0 +1,88 @@
#!/usr/bin/python3
# encoding:utf-8
import unittest
from DBConnection import DBConnection
import logging
import configparser
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
schueler = [
{'name':'Simpson','vorname': 'Bart'},
{'name':'Simpson','vorname':'Lisa'},
{'name':'van Houten','vorname':'Milhouse'},
{'name':'Wiggum','vorname':'Ralph'},
{'name':'Jones','vorname':'Jimbo'}
]
class TestDBConnection(unittest.TestCase):
@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')
@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
# input('Press Any Key')
#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)
# 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))
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])
def test_connected(self):
self.assertTrue(self.db_connection.isConnected())
def test_create_table(self):
self.assertTrue(self.db_connection.isConnected())
if __name__ =='__main__':
unittest.main()
Loading…
Cancel
Save