17 Commits

Author SHA1 Message Date
124bff5e06 Kapitel 18: Generatoren und list comprehension 2023-06-08 19:38:09 +02:00
9e78b44184 Teil17 Datenbanken 2023-05-27 16:15:35 +02:00
7f41480cc8 Datenbank über unittest. 2023-05-26 13:17:20 +02:00
67c7d20b1b working directory db 2023-05-07 10:30:07 +02:00
3598abe51f teil 16 Testing in progress. 2023-05-05 10:03:11 +02:00
a2369f3863 teil15 Aufräumarbeiten. 2023-05-05 09:48:00 +02:00
03beec6a85 Kapitel 15 GPIO 2023-05-05 06:30:12 +02:00
3434ab2d9f test_excel() implementiert. 2023-05-04 07:48:33 +02:00
995f9ff194 testing Kapitel (Vorbereitung) 2023-05-04 06:48:13 +02:00
c76aa08c83 Taster lesen. 2023-05-03 10:44:36 +02:00
63210ee88d Schalten der LED per URL-Request 2023-04-30 10:01:00 +02:00
c8b9518c11 Logging Kapitel 2023-04-29 16:32:52 +02:00
9f4428cdbc logging 2023-04-29 15:40:31 +02:00
acc0f965ff Arbeitsversion von Teil 14. 2023-04-25 13:27:20 +02:00
c8ea31042e _pycache__ gelöscht
network Package mit __init__.py ausgestattet.
2023-04-24 10:49:25 +02:00
601b5de46d teil13 2023-04-20 06:17:41 +02:00
3871bb8d99 Aufräumarbeiten 2023-04-02 14:14:53 +02:00
46 changed files with 1330 additions and 91 deletions

View File

@@ -16,4 +16,11 @@ CC-BY-SA Olli Graf
| 9 | Typkonvertierung| | 9 | Typkonvertierung|
|10 | Dateien| |10 | Dateien|
|11 | Datum und Zeit| |11 | Datum und Zeit|
|12 | Multithreading|
|13 | Netzwerk|
|14 | Logging|
|15 | GPIO|
|16 | Testing|
|17 | Datenbank|
|17 | Generatoren und list comprehension|

2
flask/.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
./__pycache__
renderhtml/__pycache__

View File

@@ -1,46 +0,0 @@
from timeit import default_timer as timer
import time
import threading
def calc_square(numbers, verbose=False):
for n in range(1,numbers):
q= n*n
if verbose:
print(f'\n{n} ^ 2 = {q}')
time.sleep(0.1)
def calc_cube(numbers,verbose=False):
for n in range(1,numbers):
k = n*n*n
if verbose:
print(f'\n{n} ^ 3 = {k}')
time.sleep(0.1)
start = timer()
thread_square = threading.Thread(target=calc_square, args=(100,True))
thread_cube = threading.Thread(target=calc_cube, args=(100,True))
thread_cube.start()
thread_square.start()
thread_cube.join()
thread_square.join()
ende = timer()
differenz_mit_print = ende - start
print(f'Zeit mit print():{differenz_mit_print}s')
start = timer()
thread_square = threading.Thread(target=calc_square, args=(100,False))
thread_cube = threading.Thread(target=calc_cube, args=(100,False))
thread_cube.start()
thread_square.start()
thread_cube.join()
thread_square.join()
ende = timer()
differenz_ohne_print = ende - start
print(f'Zeit ohne print():{differenz_ohne_print}s')

View File

@@ -1,45 +0,0 @@
import asyncio
import random
import time
gesamtwert = 0
message = ''
async def calculate_sum(lock, thread_nummer):
global gesamtwert
global message
for i in range(100 + thread_nummer):
async with lock:
zwischen = gesamtwert
#Zufällige Wartezeit zwischen 0,1s und 0,5s
wartezeit = random.randint(1,5+thread_nummer)
time.sleep(0.1 * wartezeit)
zwischen += 1
async with lock:
gesamtwert = zwischen
message = f'Thread {thread_nummer} fertig'
print(f'Thread + {thread_nummer} abgearbeitet.')
async def main():
#async Lock erzeugen
lock = asyncio.Lock()
# drei Threads starten, um die Summe der Zahlen zu berechnen
threads = [asyncio.create_task(calculate_sum(lock, _+1)) for _ in range(3)]
#Warten auf alle Threads
await asyncio.gather(*threads)
print(f'gesamtwert={gesamtwert}')
print(f'message={message}')
if __name__ == '__main__':
asyncio.run(main())

2
teil13/.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
./__pycache__
./network/__pycache__/*

Binary file not shown.

15
teil13/fib.py Normal file
View File

@@ -0,0 +1,15 @@
# fib.py
import sys
# berechnet rekursiv die Fibonaccizahl n.
def fib(n):
# Rekursionsbedingung
if n in [0,1]:
return 1
else:
# Rekursionsaufruf
return fib(n-1) + fib(n-2)
element = int(sys.argv[1])
print(fib(element))

65
teil13/fibclient.py Normal file
View File

@@ -0,0 +1,65 @@
# encoding: utf-8
# fibclient.py
import socket
import time
#from network.const import __ADDRESSREMOTE__
#from network.const import __ADDRESSLOCAL__
from network import __ADDRESSLOCAL__
DISCONNECT_MESSAGE = "!DISCONNECT"
# Verbindung herstellen
def connect():
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# für fibserver auf einem Remote-Pi
# client.connect(__ADDRESSREMOTE__)
# für fibserver auf lokalem Pi (import oben beachten!)
client.connect(__ADDRESSLOCAL__)
return client
# msg senden
def send(client, msg):
message = msg.encode('utf-8')
client.send(message)
# Daten empfangen
def receive(conn):
msg = conn.recv(64).decode('utf-8')
return msg
# Hauptmethode
def start():
# String von Console lesen
answer = input('Verbindung aufbauen? (ja/nein)? ')
if answer.lower() != 'ja':
return
# Verbindung aufbauen
connection = connect()
while True:
msg = input("Nachricht (q für quit): ")
# Bei Eingabe == 'q' Schleife beenden.
if msg == 'q':
break
# Sonst eingegeben String an Server senden.
send(connection, msg)
# und auf Ergebnis warten.
result = receive(connection)
print(f'result={result}')
# Fibonaccizahl isolieren.
fib = int(result.split('[')[0])
print(f'Fibonnaci-Zahl ist {fib}')
print(f'[RESULT] {result}')
# zum Beenden die Nachricht 'quit an den Server senden
send(connection, 'quit')
time.sleep(1)
print('Disconnected')
# Hauptprogramm: Client starten
if __name__ == '__main__':
start()

92
teil13/fibserver.py Executable file
View File

@@ -0,0 +1,92 @@
#! /usr/bin/python3
import socket
import threading
from network import __ADDRESSLOCAL__
# Server-Socket aufbauen
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#Socket an unsere Server-Adresse binden
server.bind(__ADDRESSLOCAL__)
# Set mit den verbundenen Clients
clients = set()
# Locks für den Zugriff auf das Set
clients_lock = threading.Lock()
# Unsere Arbeitsmethode: rekursive Fibonacci-Reihe
def fib(n):
# Rekursionsbedingung
if n in [0,1]:
return 1
else:
# Rekursionsaufruf
return fib(n-1) + fib(n-2)
#bearbeitet eine eingehende Connection annehmen, pro Client ein seperater Thread
def handle_connection(conn,addr):
print(f"[NEW CONNECTION] {addr} Connected")
try:
connected = True
while connected:
msg = conn.recv(64).decode('utf-8')
print(f'[RECEIVED] {msg}')
if not msg:
break
else:
if msg.startswith('fib:'):
element = int(msg[4:])
fibo = fib(element)
print(f'[SENDING] fib({element})={fibo}')
# Ergebnis an Client senden.
with clients_lock:
for c in clients:
c.sendall(f'{fibo}'.encode('utf-8'))
else:
pass
if msg == 'quit':
print('[DISCONNECT]')
connected = False
# with clients_lock:
# for c in clients:
# c.sendall(f"[{addr}] {msg}".encode('utf-8'))
finally:
# Clientverbindung abbauen und aus Set entfernen.
with clients_lock:
clients.remove(conn)
conn.close()
# Start des Servers
def start():
print('[SERVER STARTED]!')
# Horchen nach eingehender Verbindung
server.listen()
while True:
# Neue Verbindung annehmen
conn, addr = server.accept()
# Neuen Client in Set aufnehmen
with clients_lock:
print(f'[ADDING CONNECTION] {conn}')
clients.add(conn)
#Thread zur Bearbeitung des Requests starten.
thread = threading.Thread(target=handle_connection, args=(conn, addr))
thread.start()
#Hauptprogramm: Serverstart einleiten
if __name__ == '__main__':
start()

11
teil13/get_requests.py Normal file
View File

@@ -0,0 +1,11 @@
import requests
url = requests.get('https://raspithek.de')
data = url.text
print(data)

20
teil13/get_urllib.py Normal file
View File

@@ -0,0 +1,20 @@
# encoding: UTF-8
import urllib.request
# URL öffnen und Daten anfordern. Ohne spezielle Dateiangabe
# liefert der Webserver immer die index.html zurück.
fp = urllib.request.urlopen('https://raspithek.de')
# Daten auslesen
data = fp.read()
print(type(data))
# Die Daten sin vom Datentyp Bytes und müssen daher erst
# zum UTF-8 String umkodiert werden.
str = data.decode('utf8')
fp.close()
print(str)

554
teil13/index.html Normal file

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,3 @@
from .const import __ADDRESSLOCAL__
from .const import __ADDRESSREMOTE__

10
teil13/network/code.py Normal file
View File

@@ -0,0 +1,10 @@
def receive(conn):
msg = conn.recv(64).decode('utf-8')
return msg
def send(client, msg):
message = msg.encode('utf-8')
client.send(message)

19
teil13/network/const.py Normal file
View File

@@ -0,0 +1,19 @@
#network/const.py
import socket
# TCP Port
__PORT__ = 6554
#Server-IP
# für den Server (immer localhost)
__SERVERLOCAL__ = socket.gethostbyname(socket.gethostname())
#für Server auf einem Remote-Pi
__SERVERREMOTE__ = socket.gethostbyname('gabbo')
# Server-Adresse (IP,Port)
#lokal
__ADDRESSLOCAL__ = (__SERVERLOCAL__,__PORT__)
#remote
__ADDRESSREMOTE__ = (__SERVERREMOTE__,__PORT__)

3
teil13/substring.py Normal file
View File

@@ -0,0 +1,3 @@
print('fib:58'[4:])

5
teil13/test.py Normal file
View File

@@ -0,0 +1,5 @@
from network.const import __SERVER__
from network.const import __ADDRESS__
print(f'Server:{__SERVER__}')
print(f'Address:{__ADDRESS__}')

2
teil14/.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
./__pycache__
./network/__pycache__/*

4
teil14/conf.py Normal file
View File

@@ -0,0 +1,4 @@
import logging
__LOGLEVEL__ = logging.DEBUG

5
teil14/test.log Normal file
View File

@@ -0,0 +1,5 @@
2023-04-29 13:11:26,397 [DEBUG] testlogging: DEBUG/Drucken wir vorbereitet
2023-04-29 13:11:26,397 [INFO] testlogging: INFO/Es werden 2 Seiten gedruckt
2023-04-29 13:11:26,397 [WARNING] testlogging: WARN/nicht genügend Papier im Drucker
2023-04-29 13:11:26,397 [ERROR] testlogging: ERROR/Drucker nicht gefunden.
2023-04-29 13:11:26,397 [CRITICAL] testlogging: CRITICAL/Drucker brennt

17
teil14/testlogging.py Normal file
View File

@@ -0,0 +1,17 @@
import logging
from conf import __LOGLEVEL__
logging.basicConfig( format='%(asctime)s [%(levelname)s] %(funcName)s: %(message)s',
# filename='test.log',
level=__LOGLEVEL__)
def testlogging():
print('testlogging() aufgerufen.')
logging.debug('DEBUG/Druckdaten werden gesendet.')
logging.info('INFO/Es werden 2 Seiten gedruckt')
logging.warning('WARN/nicht genügend Papier im Drucker')
logging.error('ERROR/Drucker nicht gefunden.')
logging.critical('CRITICAL/Drucker brennt')
if __name__ == '__main__':
testlogging()

2
teil15/.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
./__pycache__
./network/__pycache__/*

37
teil15/button.py Executable file
View File

@@ -0,0 +1,37 @@
#! /usr/bin/python3
import RPi.GPIO as GPIO
from time import sleep
import logging
__PIN__ = 16 # GPIO Pin, für den Taster
__WAIT__ = 0.5 # Warten für 0,5 Sekunden
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
if __name__ =='__main__':
GPIO.setwarnings(False)
# benutze Broadcom Pin Nummerierung
GPIO.setmode(GPIO.BCM)
logging.debug(f'konfiguriere Pin ${__PIN__} als Input.')
# Pin 16 als Input mit Pull-Up-Widerstand schalten.
GPIO.setup(__PIN__, GPIO.IN, pull_up_down=GPIO.PUD_UP)
pressed = False
logging.debug('Start der Schleife.')
try:
while True:
if not GPIO.input(__PIN__):
if not pressed:
logging.info('Button wurde gedrückt.')
pressed = True
else:
logging.debug('Button nicht gedrückt.')
pressed = False
sleep(__WAIT__)
except KeyboardInterrupt:
logging.debug('Abbruch durch Benutzer.')
finally:
pass

50
teil15/led.py Executable file
View File

@@ -0,0 +1,50 @@
#! /usr/bin/python3
import RPi.GPIO as GPIO
from time import sleep
__PIN__ = 23 # GPIO Pin, den wir nutzen
__WAIT__ = 0.5 # Warten für 0,5 Sekunden
GPIO.setwarnings(False)
# benutze Broadcom Pin Nummerierung
GPIO.setmode(GPIO.BCM)
# Pin 23 als Output schalten.
GPIO.setup(__PIN__,GPIO.OUT)
#Zustand der LED setzen
def setLED(state):
GPIO.output(__PIN__,state)
# Dauerschleife, die die LED im Wechsel ein- und ausschaltet.
# Wird gestoppt mit CTRL-C
def blinkloop():
try:
while True:
# Pin auf HIGH setzen schaltet die LED ein
setLED(GPIO.HIGH)
print('LED ein')
sleep(__WAIT__)
# Pin auf LOW setzen schaltet die LED aus
setLED(GPIO.LOW)
print('LED aus')
sleep(__WAIT__)
except KeyboardInterrupt:
pass
finally:
# Zum schluss immer die LED ausschalten.
setLED(GPIO.LOW)
if __name__ =='__main__':
blinkloop()
GPIO.cleanup(__PIN__)

69
teil15/ledserver.py Executable file
View File

@@ -0,0 +1,69 @@
#! /usr/bin/python3
from http.server import SimpleHTTPRequestHandler
import socketserver
import logging
from led import setLED
responsehtml = '<html><body> <h3>LED-Server</h3><p> LED ist $ledstate_str</p></body></html>'
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
# Da http ein stateless Protokoll ist, wird der Handler bei jedem Request neu
# aufgebaut! Deswegen benutzen wir hier setup() statt __init__()
class LEDHttpRequestHandler(SimpleHTTPRequestHandler):
def setup(self):
logging.debug('setting up Handler')
self.ledstate = False
super().setup()
def do_GET(self):
logging.debug(f'path= {self.path}')
logging.debug('GET: empfangen')
self.send_response(200)
self.send_header('Content-Type:','text/html')
# Ende der Header markieren, dass ist wichtig, sonst kommt der Response nicht zurück!
self.end_headers()
#Wenn im Requestpath 'led='on'enthalten ist, schalten wir die LED ein. Bei off aus
if '?led=' in self.path:
if 'on' in self.path:
logging.debug('schalte LED ein')
self.ledstate = True
setLED(self.ledstate)
elif 'off' in self.path:
logging.debug('schalte LED aus')
self.ledstate = False
setLED(self.ledstate)
logging.debug(f'ledstate={self.ledstate}')
response = responsehtml.replace('$ledstate_str',('aus','an')[self.ledstate])
self.wfile.write(response.encode())
class http_server:
def __init__(self):
def handler(*args):
myHandler(*args)
server = HTTPServer(('', 8080), handler)
if __name__ == '__main__':
logging.debug('MAIN: start')
setLED(False) # Beim Start LED ausschalten
#Request-Handler aufbauen
handler = LEDHttpRequestHandler
handler.ledstate = False
#Socketserver mit unserem Handler starten
with socketserver.TCPServer(('',8080),handler) as httpd:
logging.info('SERVER: start')
try:
#Server läuft bis CTRL-C gedrückt wird.
httpd.serve_forever()
except KeyboardInterrupt:
pass
finally:
setLED(False)

View File

@@ -0,0 +1,3 @@
from .const import __ADDRESSLOCAL__
#from .const import __ADDRESSREMOTE__

10
teil15/network/code.py Normal file
View File

@@ -0,0 +1,10 @@
def receive(conn):
msg = conn.recv(64).decode('utf-8')
return msg
def send(client, msg):
message = msg.encode('utf-8')
client.send(message)

19
teil15/network/const.py Normal file
View File

@@ -0,0 +1,19 @@
#network/const.py
import socket
# TCP Port
__PORT__ = 6554
#Server-IP
# für den Server (immer localhost)
__SERVERLOCAL__ = socket.gethostbyname(socket.gethostname())
#für Server auf einem Remote-Pi
#__SERVERREMOTE__ = socket.gethostbyname('gabbo')
# Server-Adresse (IP,Port)
#lokal
__ADDRESSLOCAL__ = (__SERVERLOCAL__,__PORT__)
#remote
#__ADDRESSREMOTE__ = (__SERVERREMOTE__,__PORT__)

2
teil16/.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
./__pycache__
./network/__pycache__/*

BIN
teil16/Fib-Testdaten.xlsx Normal file

Binary file not shown.

13
teil16/fib.py Normal file
View File

@@ -0,0 +1,13 @@
# fib.py
import sys
# berechnet rekursiv die Fibonaccizahl n.
def fib(n):
# Rekursionsbedingung
if n in [0,1]:
return 1
else:
# Rekursionsaufruf
return fib(n-1) + fib(n-2)

20
teil16/testfib.py Normal file
View File

@@ -0,0 +1,20 @@
from fib import fib
import logging
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
logging.debug('Test startet')
result5 = fib(5)
result8 = fib(8)
if result5 == 8 and result8 == 34:
logging.info('Tests erfolgreich')
else:
logging.error('Tests fehlerhaft')
logging.debug('Testende')

57
teil16/unittestfib.py Normal file
View File

@@ -0,0 +1,57 @@
# coding: utf-8
import unittest
import numbers
from fib import fib
import logging
import pandas as pd
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
class TestFib(unittest.TestCase):
# Methode zur Division a/b.
def divide(self, a,b):
if b == 0:
raise ValueError('Dividend darf nicht 0 sein.')
return int(a/b)
# Die setUp() Methode wird zu Beginn jedes Testcases aufgrufen
def setUp(self):
logging.debug('setting up test')
self.results = [1,1,2,3,5,8,13,21,34,55]
#die beiden Tests aus ./testfib.py mit unittest
def test_fib(self):
# assertEqual(n,r) testet, ob das Resultat r der Vorgabe n entspricht.
self.assertEqual(8,fib(5))
self.assertEqual(34,fib(8))
def test_mult_fib(self):
results =[1,1,2,3,5,8,13,21,34,55]
for n in range(0,9):
self.assertEqual(results[n],fib(n))
def test_divide(self):
self.assertEqual(2, self.divide(2,1))
self.assertEqual(2, self.divide(4,2))
self.assertEqual(3, self.divide(6,2))
with self.assertRaises(ValueError):
self.divide(1,0)
def test_excel(self):
data = pd.read_excel('./Fib-Testdaten.xlsx')
results = data["Ergebnis"]
for key, result in data.items():
if isinstance(key, numbers.Number): #Überspringen der Überschriftenzeile
self.assertEqual(result, fib(int(key)))
if __name__ == '__main__':
unittest.main()

3
teil17/.gitignore vendored Normal file
View File

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

45
teil17/DBConnection.py Normal file
View File

@@ -0,0 +1,45 @@
#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.

Binary file not shown.

9
teil17/connectdb.py Normal file
View File

@@ -0,0 +1,9 @@
from DBConnection import DBConnection
import logging
logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
db_connection = DBConnection()
db_connection.connect('database','piuser','SW_Ep.6:ROTJ','pidb')
logging.debug(f'Datenbankverbindung aufgebaut ${db_connection.isConnected()}')

8
teil17/db.ini.sample Normal file
View File

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

88
teil17/testdb.py Executable file
View File

@@ -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()

1
teil18/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
./__pycache__

15
teil18/fib_gen.py Normal file
View File

@@ -0,0 +1,15 @@
import sys
def fib_generator():
a,b = 0,1
while True:
yield a
a,b = b, a+b
fibo = fib_generator()
for _ in range(int(sys.argv[1])):
print(next(fibo))

6
teil18/squares_for.py Normal file
View File

@@ -0,0 +1,6 @@
squares = []
for x in range(1,6):
squares.append(x**2)
print(squares)

View File

@@ -0,0 +1,3 @@
squares = [x **2 for x in range(1,6)]
print(squares)

11
teil18/vornamen_filter.py Normal file
View File

@@ -0,0 +1,11 @@
schueler = [
{'name':'Simpson','vorname': 'Bart'},
{'name':'Simpson','vorname':'Lisa'},
{'name':'van Houten','vorname':'Milhouse'},
{'name':'Wiggum','vorname':'Ralph'},
{'name':'Jones','vorname':'Jimbo'}
]
vornamen = [s['vorname'] for s in schueler if s['name'] == 'Simpson']
print(vornamen)

4
teil18/zahlen_for.py Normal file
View File

@@ -0,0 +1,4 @@
zahlen = [1,2,3,4,5,6,7,8,9,10]
for n in zahlen:
print(n)

19
teil18/zahlen_gen.py Normal file
View File

@@ -0,0 +1,19 @@
def gen_zahl():
n = 0
while n < 10:
n += 1
yield n
zahlen = gen_zahl()
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))
print(next(zahlen))