+ Im echten Alltag eines Software-Entwicklers oder Data Engineers sind 80% aller Rohdaten unstrukturiert: E-Mails, Server-Logs, Tweets, Rechnungen oder Kundeneingaben.
+ Einfache String-Methoden wie .find() oder .split() reichen oft nicht aus.
+
+
+
+
🕵️ Die Detektiv-Analogie mit der Suchlupe:
+
+ Ein regulärer Ausdruck ist wie ein Phantombild:
+ Statt nach einem festen Text zu suchen („Suche exakt nach 'Peter'“), suchst du nach einer Struktur:
+
+ „Suche 2 Großbuchstaben, gefolgt von 2 Ziffern, gefolgt von weiteren Zahlen – egal welche Nummern es genau sind, das ist eine IBAN!“
+
+
+
+
+
+
+
2. Cheat-Sheet: Die wichtigsten Metazeichen & Quantifizierer
+
+ In Python nutzen wir das Modul import re. Regex-Muster werden als Raw-Strings (r"...") definiert:
+
+
+
+
+
+
Zeichen
+
Kategorie
+
Bedeutung
+
Beispiel
+
+
+
+
+
\d
+
Metazeichen
+
Beliebige Ziffer (0-9)
+
r"\d{4}" → "2026"
+
+
+
\w
+
Metazeichen
+
Wortzeichen (Buchstabe, Ziffer, Unterstrich)
+
r"#\w+" → "#python"
+
+
+
\s
+
Metazeichen
+
Whitespace (Leerzeichen, Tab, Umbruch)
+
r"\s+"
+
+
+
.
+
Metazeichen
+
Beliebiges Zeichen (außer Zeilenumbruch)
+
r"a.c" → "abc", "a9c"
+
+
+
+
+
Quantifizierer
+
Mindestens 1 Mal (1 bis unendlich)
+
r"\d+" → "42"
+
+
+
*
+
Quantifizierer
+
0 oder mehrmals (optional beliebig oft)
+
r"ab*c" → "ac", "abc"
+
+
+
?
+
Quantifizierer
+
Optional (0 oder 1 Mal)
+
r"https?" → "http", "https"
+
+
+
{min,max}
+
Quantifizierer
+
Von min bis max Wiederholungen
+
r"\d{2,4}"
+
+
+
^ und $
+
Anker
+
Zeilenanfang (^) und Zeilenende ($)
+
r"^\d+$" (nur Ziffern)
+
+
+
+
+
+
+
+
3. Die 4 Haupt-Methoden in Python
+
+
+
+
+
Methode
+
Zweck
+
Code-Beispiel
+
+
+
+
+
re.search()
+
Sucht den ersten Treffer irgendwo im String.
+
m = re.search(r"\d+", "Raum 404")
+
+
+
re.fullmatch()
+
Prüft, ob der gesamte String dem Muster entspricht (Validierung).
+
re.fullmatch(r"^[a-z]+@[a-z]+\.[a-z]{2,}$", mail)
+
+
+
re.findall()
+
Sammelt alle Vorkommnisse in einer Python-Liste.
+
re.findall(r"#\w+", tweet)
+
+
+
re.sub()
+
Sucht Treffer und ersetzt sie (auch per Lambda/Funktion).
+
re.sub(r"\d", "*", "Pin: 1234") → "Pin: ****"
+
+
+
+
+
+
💡 Benannte Gruppen mit (?P<name>...):
+
Mit benannten Gruppen kannst du Daten direkt in ein lesbares Dictionary extrahieren:
+ In Python gibt es oft zwei Wege, eine Aufgabe zu lösen: Den langen, imperativen Weg mit vielen Hilfsvariablen und verschachtelten Schleifen – oder den eleganten, funktionalen Weg (Pythonic Code).
+
+
+
+
🏭 Die automatisierte Fließband-Analogie:
+
+ Stelle dir eine moderne Fertigungsstraße vor:
+
+
+
enumerate() (Die Etikettierstation): Klebt an jedes Teil automatisch die aktuelle Positionsnummer.
+
zip() (Die Fügestation): Bringt Bauteile von zwei parallelen Fließbändern paarweise zusammen.
+
filter() (Die Qualitätsweiche): Lässt nur Bauteile passieren, die ein Prüfkriterium erfüllen.
+
map() (Die Veredelungsstation): Wendet eine Transformation auf jedes durchlaufende Teil an.
+
all() / any() (Die Sensorüberwachung): Meldet sofort, ob alle Teile fehlerfrei sind oder mindestens ein Alarm anliegt.
+
+
+
+
+
+
+
2. Cheat-Sheet: Funktionale Idiome auf einen Blick
+
+
+
+
+
Funktion / Idiom
+
Zweck
+
Typisches Code-Beispiel
+
Ergebnis
+
+
+
+
+
enumerate(l, start)
+
Iteriert gleichzeitig über Index und Element.
+
[f"{i}. {x}" for i, x in enumerate(["A","B"], 1)]
+
['1. A', '2. B']
+
+
+
zip(a, b)
+
Verknüpft mehrere Sequenzen paarweise.
+
dict(zip(["Name", "Alter"], ["Max", 25]))
+
{'Name': 'Max', 'Alter': 25}
+
+
+
lambda x: ...
+
Anonyme Mini-Funktion als Einzeiler.
+
verdoppeln = lambda x: x * 2
+
verdoppeln(5) → 10
+
+
+
map(f, iter)
+
Wendet Funktion f auf alle Elemente an.
+
list(map(lambda x: x**2, [1, 2, 3]))
+
[1, 4, 9]
+
+
+
filter(p, iter)
+
Filtert Elemente nach Prädikat p.
+
list(filter(lambda x: x > 0, [-2, 0, 5]))
+
[5]
+
+
+
sorted(..., key=...)
+
Sortiert nach beliebigem Schlüssel (Lambda).
+
sorted(user, key=lambda u: u["alter"])
+
Sortierte Liste
+
+
+
all(iter)
+
Gibt True, wenn ALLE Elemente wahr sind.
+
all(x >= 18 for x in [20, 25, 18])
+
True
+
+
+
any(iter)
+
Gibt True, wenn MINDESTENS EIN Element wahr ist.
+
any(x == 0 for x in [1, 2, 0, 4])
+
True
+
+
+
+
+
+
+
+
3. 📋 Deine Aufgaben-Checkliste
+
Öffne die Datei aufgabe.py und löse die 6 Kern-Aufgaben:
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
💡 Lösungshinweise (Aufklappen bei Bedarf)
+
+
+ 🔍 Tipp zu TODO 3: Map & Filter verketten
+
+
Du kannst filter() direkt in map() schachteln und mit list() materialisieren:
+
+
returnlist(map(lambdax: x ** 2, filter(lambdax: x % 2 != 0, zahlen)))
+
+
+
+
+
+ 🔍 Tipp zu TODO 4: Sortieren mit Lambda-Key
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/lehrpfad_4_professional_data_engineering/18_funktionale_programmierung_lambda/musterloesung.py b/lehrpfad_4_professional_data_engineering/18_funktionale_programmierung_lambda/musterloesung.py
new file mode 100644
index 0000000..fa6a59b
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/18_funktionale_programmierung_lambda/musterloesung.py
@@ -0,0 +1,76 @@
+"""
+Kapitel 18: Funktionale Programmierung & Pythonic Code – Musterlösung 💡
+========================================================================
+Vollständige Referenzlösung für alle Teilaufgaben aus Kapitel 18.
+"""
+
+from typing import List, Dict, Any, Optional
+
+
+def nummerierte_liste(elemente: List[str], start: int = 1) -> List[str]:
+ """
+ Erstellt eine nummerierte Liste von Strings mit enumerate().
+ """
+ return [f"{i}. {item}" for i, item in enumerate(elemente, start=start)]
+
+
+def kombiniere_katalog(artikel: List[str], preise: List[float]) -> Dict[str, float]:
+ """
+ Verknüpft Artikelnamen und Preise mittels zip() zu einem Dictionary.
+ """
+ return dict(zip(artikel, preise))
+
+
+def quadriere_und_filtere(zahlen: List[int]) -> List[int]:
+ """
+ Filtert ungerade Zahlen und quadriert sie funktional mit map() und filter().
+ """
+ return list(map(lambda x: x ** 2, filter(lambda x: x % 2 != 0, zahlen)))
+
+
+def sortiere_personen_nach_alter(personen: List[Dict[str, Any]], absteigend: bool = False) -> List[Dict[str, Any]]:
+ """
+ Sortiert eine Liste von Personen-Dictionaries anhand von 'alter' mit sorted() und key=lambda.
+ """
+ return sorted(personen, key=lambda p: p["alter"], reverse=absteigend)
+
+
+def alle_volljaehrig(alter_liste: List[int], mindestalter: int = 18) -> bool:
+ """
+ Prüft mit all(), ob alle Personen in der Liste mindestens mindestalter Jahre alt sind.
+ """
+ return all(alter >= mindestalter for alter in alter_liste)
+
+
+def mindestens_ein_treffer(ergebnisse: List[bool]) -> bool:
+ """
+ Prüft mit any(), ob mindestens ein Element True ist.
+ """
+ return any(ergebnisse)
+
+
+def finde_extrem_produkt(produkte: List[Dict[str, Any]], modus: str = "teuerstes") -> Dict[str, Any]:
+ """
+ Findet das teuerste oder günstigste Produkt mit max() / min() und key=lambda.
+ """
+ if not produkte:
+ raise ValueError("Produktliste darf nicht leer sein!")
+ if modus == "teuerstes":
+ return max(produkte, key=lambda p: p["preis"])
+ elif modus == "billigstes":
+ return min(produkte, key=lambda p: p["preis"])
+ else:
+ raise ValueError(f"Ungültiger Modus '{modus}'. Erlaubt sind 'teuerstes' und 'billigstes'.")
+
+
+if __name__ == "__main__":
+ print("Musterlösung Kapitel 18:")
+ print("1. Enumerate:", nummerierte_liste(["Python", "Rust", "Go"]))
+ print("2. Zip:", kombiniere_katalog(["Kaffee", "Kuchen"], [3.5, 4.2]))
+ print("3. Map/Filter:", quadriere_und_filtere([1, 2, 3, 4, 5]))
+ team = [{"name": "A", "alter": 30}, {"name": "B", "alter": 20}]
+ print("4. Custom Sort:", sortiere_personen_nach_alter(team))
+ print("5. all():", alle_volljaehrig([20, 21, 18]))
+ print("6. any():", mindestens_ein_treffer([False, True]))
+ items = [{"titel": "A", "preis": 10}, {"titel": "B", "preis": 50}]
+ print("7. Extremwert:", finde_extrem_produkt(items, "teuerstes"))
diff --git a/lehrpfad_4_professional_data_engineering/18_funktionale_programmierung_lambda/test_aufgabe.py b/lehrpfad_4_professional_data_engineering/18_funktionale_programmierung_lambda/test_aufgabe.py
new file mode 100644
index 0000000..20169d5
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/18_funktionale_programmierung_lambda/test_aufgabe.py
@@ -0,0 +1,182 @@
+import unittest
+import sys
+from pathlib import Path
+
+# Sicherstellen, dass das aktuelle Verzeichnis im Python-Pfad liegt
+sys.path.insert(0, str(Path(__file__).parent))
+
+try:
+ import aufgabe
+except ImportError as e:
+ raise ImportError(f"Konnte 'aufgabe.py' nicht importieren: {e}")
+
+
+class TestKapitel18(unittest.TestCase):
+ """
+ Testsuite für Kapitel 18: Funktionale Programmierung & Pythonic Code.
+ """
+
+ def setUp(self):
+ self.nummerierte_liste = getattr(aufgabe, "nummerierte_liste", None)
+ self.kombiniere_katalog = getattr(aufgabe, "kombiniere_katalog", None)
+ self.quadriere_und_filtere = getattr(aufgabe, "quadriere_und_filtere", None)
+ self.sortiere_personen_nach_alter = getattr(aufgabe, "sortiere_personen_nach_alter", None)
+ self.alle_volljaehrig = getattr(aufgabe, "alle_volljaehrig", None)
+ self.mindestens_ein_treffer = getattr(aufgabe, "mindestens_ein_treffer", None)
+ self.finde_extrem_produkt = getattr(aufgabe, "finde_extrem_produkt", None)
+
+ self.assertIsNotNone(self.nummerierte_liste, "Funktion 'nummerierte_liste' nicht gefunden!")
+ self.assertIsNotNone(self.kombiniere_katalog, "Funktion 'kombiniere_katalog' nicht gefunden!")
+ self.assertIsNotNone(self.quadriere_und_filtere, "Funktion 'quadriere_und_filtere' nicht gefunden!")
+ self.assertIsNotNone(self.sortiere_personen_nach_alter, "Funktion 'sortiere_personen_nach_alter' nicht gefunden!")
+ self.assertIsNotNone(self.alle_volljaehrig, "Funktion 'alle_volljaehrig' nicht gefunden!")
+ self.assertIsNotNone(self.mindestens_ein_treffer, "Funktion 'mindestens_ein_treffer' nicht gefunden!")
+
+ # ==========================================================================
+ # 1. Enumerate – Nummerierte Listen
+ # ==========================================================================
+ def test_01_nummerierte_liste(self):
+ """Prüft enumerate() mit Standardstartwert 1 und benutzerdefiniertem Start."""
+ sprachen = ["Python", "Java", "Rust"]
+ ergebnis = self.nummerierte_liste(sprachen)
+ self.assertEqual(ergebnis, ["1. Python", "2. Java", "3. Rust"])
+
+ # Startwert 0
+ ergebnis_0 = self.nummerierte_liste(["A", "B", "C"], start=0)
+ self.assertEqual(ergebnis_0, ["0. A", "1. B", "2. C"])
+
+ # Startwert 10
+ ergebnis_10 = self.nummerierte_liste(["Erster", "Zweiter"], start=10)
+ self.assertEqual(ergebnis_10, ["10. Erster", "11. Zweiter"])
+
+ # Leere Liste
+ self.assertEqual(self.nummerierte_liste([]), [])
+
+ # ==========================================================================
+ # 2. Zip – Katalog-Kombination
+ # ==========================================================================
+ def test_02_kombiniere_katalog(self):
+ """Prüft zip() zur Kombination von Artikeln und Preisen."""
+ artikel = ["Apfel", "Banane", "Orange"]
+ preise = [0.99, 1.49, 1.99]
+ katalog = self.kombiniere_katalog(artikel, preise)
+ self.assertEqual(katalog, {"Apfel": 0.99, "Banane": 1.49, "Orange": 1.99})
+
+ # Unterschiedliche Längen (zip schneidet ab)
+ katalog_kurz = self.kombiniere_katalog(["Kaffee", "Tee"], [3.50, 2.80, 9.99])
+ self.assertEqual(katalog_kurz, {"Kaffee": 3.50, "Tee": 2.80})
+
+ # Leere Listen
+ self.assertEqual(self.kombiniere_katalog([], []), {})
+
+ # ==========================================================================
+ # 3. Map & Filter & Lambda – Pipeline
+ # ==========================================================================
+ def test_03_quadriere_und_filtere(self):
+ """Prüft Filterung ungerader Zahlen und deren Quadrierung."""
+ # Gemischte Zahlen
+ zahlen = [1, 2, 3, 4, 5, 6, 7]
+ ergebnis = self.quadriere_und_filtere(zahlen)
+ self.assertEqual(ergebnis, [1, 9, 25, 49])
+
+ # Nur gerade Zahlen -> Leeres Ergebnis
+ self.assertEqual(self.quadriere_und_filtere([2, 4, 6, 8]), [])
+
+ # Negative ungerade Zahlen: (-3)^2 = 9
+ self.assertEqual(self.quadriere_und_filtere([-3, -2, -1, 0, 1]), [9, 1, 1])
+
+ # Leere Liste
+ self.assertEqual(self.quadriere_und_filtere([]), [])
+
+ # ==========================================================================
+ # 4. Custom Sorting mit Key=Lambda
+ # ==========================================================================
+ def test_04_sortiere_personen_nach_alter(self):
+ """Prüft benutzerdefiniertes Sortieren nach Alter (aufsteigend & absteigend)."""
+ personen = [
+ {"name": "Charlie", "alter": 45},
+ {"name": "Alice", "alter": 30},
+ {"name": "Bob", "alter": 22},
+ {"name": "Dora", "alter": 18}
+ ]
+
+ # Aufsteigend
+ sortiert_auf = self.sortiere_personen_nach_alter(personen, absteigend=False)
+ namen_auf = [p["name"] for p in sortiert_auf]
+ self.assertEqual(namen_auf, ["Dora", "Bob", "Alice", "Charlie"])
+
+ # Absteigend
+ sortiert_ab = self.sortiere_personen_nach_alter(personen, absteigend=True)
+ namen_ab = [p["name"] for p in sortiert_ab]
+ self.assertEqual(namen_ab, ["Charlie", "Alice", "Bob", "Dora"])
+
+ # Original-Liste darf nicht verändert werden (neue Liste zurückgeben)
+ self.assertIsNot(sortiert_auf, personen)
+
+ # ==========================================================================
+ # 5. all() – Volljährigkeit
+ # ==========================================================================
+ def test_05_alle_volljaehrig(self):
+ """Prüft all() mit Standardmindestalter 18 und dynamischem Mindestalter."""
+ # Alle >= 18
+ self.assertTrue(self.alle_volljaehrig([18, 19, 25, 50]))
+
+ # Mindestens einer < 18
+ self.assertFalse(self.alle_volljaehrig([18, 17, 25, 50]))
+ self.assertFalse(self.alle_volljaehrig([0]))
+
+ # Anderes Mindestalter (z.B. 21)
+ self.assertTrue(self.alle_volljaehrig([21, 25, 30], mindestalter=21))
+ self.assertFalse(self.alle_volljaehrig([20, 25, 30], mindestalter=21))
+
+ # Leere Liste ist trivial True in Python
+ self.assertTrue(self.alle_volljaehrig([]))
+
+ # ==========================================================================
+ # 6. any() – Mindestens ein Treffer
+ # ==========================================================================
+ def test_06_mindestens_ein_treffer(self):
+ """Prüft any() auf Wahrheitswerte."""
+ self.assertTrue(self.mindestens_ein_treffer([False, False, True, False]))
+ self.assertTrue(self.mindestens_ein_treffer([True, True, True]))
+ self.assertFalse(self.mindestens_ein_treffer([False, False, False]))
+ self.assertFalse(self.mindestens_ein_treffer([]))
+
+ # ==========================================================================
+ # 7. Bonus: Extremwert-Finder
+ # ==========================================================================
+ def test_07_finde_extrem_produkt(self):
+ """Prüft das Finden von Extremwerten mit max/min und key=lambda."""
+ if self.finde_extrem_produkt is None:
+ return
+
+ produkte = [
+ {"titel": "Kugelschreiber", "preis": 2.50},
+ {"titel": "Laptop", "preis": 1299.00},
+ {"titel": "Kaffeetasse", "preis": 8.90},
+ {"titel": "Monitor", "preis": 349.00}
+ ]
+
+ try:
+ teuerstes = self.finde_extrem_produkt(produkte, modus="teuerstes")
+ if teuerstes is not None:
+ self.assertEqual(teuerstes["titel"], "Laptop")
+ self.assertEqual(teuerstes["preis"], 1299.00)
+
+ billigstes = self.finde_extrem_produkt(produkte, modus="billigstes")
+ if billigstes is not None:
+ self.assertEqual(billigstes["titel"], "Kugelschreiber")
+ self.assertEqual(billigstes["preis"], 2.50)
+
+ # Fehlerfälle
+ with self.assertRaises(ValueError):
+ self.finde_extrem_produkt([], modus="teuerstes")
+
+ with self.assertRaises(ValueError):
+ self.finde_extrem_produkt(produkte, modus="ungueltig")
+ except NotImplementedError:
+ pass
+
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)
diff --git a/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/README.md b/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/README.md
new file mode 100644
index 0000000..0b98003
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/README.md
@@ -0,0 +1,147 @@
+# Kapitel 19: Generatoren, Iteratoren & itertools (Memory Efficiency & Streaming) 🌊⚡
+
+Willkommen zu **Kapitel 19** im Lehrpfad 4 (*Professional Data Engineering*)!
+In diesem Modul lernst du, wie professionelle Software-Entwickler riesige Datenmengen speichereffizient im RAM verarbeiten. Statt Gigabytes an Daten auf einmal in eine Liste zu laden und dein System zum Absturz zu bringen, streamst du Werte **on-demand** Tropfen für Tropfen mit **Generatoren** und dem mächtigen **`itertools`**-Modul.
+
+---
+
+## 🧭 Didaktischer Hintergrund
+
+In Data Pipelines, Web-Servern und Big-Data-Anwendungen stößt klassisches Laden von Daten schnell an physikalische RAM-Grenzen:
+- **Speicherüberlauf (Out of Memory):** Eine Liste mit 50 Millionen Objekten belegt mehrere Gigabytes Arbeitsspeicher.
+- **Lazy Evaluation (Verzögerte Auswertung):** Werte werden erst in dem exakten Mikrosekunden-Moment berechnet, in dem sie gebraucht werden.
+- **Null RAM-Verschwendung:** Ein Generator verbraucht unabhängig davon, ob er 10 oder 10 Milliarden Elemente liefert, immer nur eine winzige, konstante Speichermenge (ca. 100 Bytes)!
+
+---
+
+## 🚰 1. Die Badewannen- vs. Wasserhahn-Analogie
+
+```
+🔴 DIE BADEWANNE (Klassische Listen & Arrays)
+┌────────────────────────────────────────────────────────┐
+│ [Wert 1, Wert 2, Wert 3, ..., Wert 10.000.000] │
+│ -> 800 MB RAM belegt! Alles muss VORHER im RAM sein! │
+└────────────────────────────────────────────────────────┘
+
+🟢 DER WASSERHAHN (Generatoren & Streams mit yield)
+┌────────────────────────────────────────────────────────┐
+│ [Wasserhahn] ── Tropfen 1 ──> [Verarbeitung] │
+│ ── Tropfen 2 ──> [Verarbeitung] │
+│ ── Tropfen 3 ──> [Verarbeitung] │
+│ -> ~100 Bytes RAM! Jeder Wert wird nach Gebrauch │
+│ wieder verworfen oder weitergeleitet. │
+└────────────────────────────────────────────────────────┘
+```
+
+---
+
+## 🔄 2. Iterables vs. Iterators (`iter()`, `next()`, `StopIteration`)
+
+In Python gibt es einen fundamentalen Unterschied:
+
+| Begriff | Was ist das? | Wie wird es erzeugt? | Typische Beispiele |
+| :--- | :--- | :--- | :--- |
+| **Iterable** | Eine Sammlung, über die man iterieren kann (besitzt `__iter__()`). | Listen, Tupel, Strings, Dicts | `[1, 2, 3]`, `"Hallo"` |
+| **Iterator** | Das Zustandsobjekt, das den aktuellen Zeiger hält und mit `next()` den nächsten Wert liefert (besitzt `__next__()`). | `it = iter(iterable)` | `iter([1, 2, 3])` |
+
+### Was passiert bei `for x in sammlung:` unter der Haube?
+1. Python ruft `it = iter(sammlung)` auf, um einen Iterator zu erhalten.
+2. In jeder Schleifenrunde wird `x = next(it)` aufgerufen.
+3. Wenn keine Elemente mehr vorhanden sind, löst Python intern die `StopIteration`-Exception aus und beendet die Schleife sauber.
+
+```python
+daten = [10, 20, 30]
+it = iter(daten)
+
+print(next(it)) # -> 10
+print(next(it)) # -> 20
+print(next(it)) # -> 30
+# print(next(it)) # -> Löst StopIteration aus!
+```
+
+---
+
+## 🪄 3. Eigene Generatoren mit `yield`
+
+Das Schlüsselwort **`yield`** verwandelt eine gewöhnliche Funktion in eine Generator-Funktion.
+- Ein `return` beendet die Funktion und vernichtet alle lokalen Variablen.
+- Ein `yield` pausiert die Funktion, gibt den Zwischenwert zurück und merkt sich den exakten Programmzustand (Zeile, lokale Variablen), um beim nächsten `next()` genau dort weiterzumachen!
+
+```python
+def mein_generator():
+ print("Starte Schritt 1...")
+ yield "Erster Wert"
+ print("Mache weiter bei Schritt 2...")
+ yield "Zweiter Wert"
+ print("Fertig!")
+
+gen = mein_generator() # Noch kein Code wird ausgeführt!
+print(next(gen)) # Druckt: Starte Schritt 1... -> Erster Wert
+print(next(gen)) # Druckt: Mache weiter... -> Zweiter Wert
+```
+
+### Unendliche Generatoren:
+Mit `yield` sind unendliche Sequenzen möglich, ohne dass der Speicher überläuft:
+
+```python
+def endlos_zaehler(start=0, schritt=1):
+ aktuell = start
+ while True:
+ yield aktuell
+ aktuell += schritt
+```
+
+---
+
+## ⚡ 4. Generator-Expressions: `(x**2 for x in ...)`
+
+Genauso wie es List Comprehensions `[x for x in ...]` gibt, kannst du mit runden Klammern `(x for x in ...)` eine schlanke **Generator Expression** erzeugen:
+
+```python
+# 🔴 List Comprehension: Erzeugt 10 Mio. Zahlen sofort im RAM (ca. 80 MB!)
+quadrate_liste = [x**2 for x in range(10_000_000)]
+
+# 🟢 Generator Expression: Kostet ca. 100 Bytes RAM!
+quadrate_gen = (x**2 for x in range(10_000_000))
+
+print(next(quadrate_gen)) # 0
+print(next(quadrate_gen)) # 1
+print(next(quadrate_gen)) # 4
+```
+
+---
+
+## 🧰 5. Das `itertools`-Modul (Die High-Performance Toolbox)
+
+Pythons Standardbibliothek liefert mit `import itertools` hochgradig optimierte C-Iteratoren:
+
+| Funktion | Erklärung | Code-Beispiel |
+| :--- | :--- | :--- |
+| `itertools.chain(*iterables)` | Hängt mehrere Iterables lückenlos hintereinander | `chain([1, 2], [3, 4])` $\rightarrow$ `1, 2, 3, 4` |
+| `itertools.cycle(iterable)` | Wiederholt ein Iterable in einer Endlosschleife | `cycle(['A', 'B'])` $\rightarrow$ `A, B, A, B, A...` |
+| `itertools.count(start, step)` | Unendlicher Zähler | `count(10, 2)` $\rightarrow$ `10, 12, 14, 16...` |
+| `itertools.permutations(p, r)` | Alle geordneten Anordnungen der Länge $r$ | `permutations('AB', 2)` $\rightarrow$ `('A','B'), ('B','A')` |
+| `itertools.combinations(p, r)` | Alle ungeordneten Kombinationen ohne Duplikate | `combinations('ABC', 2)` $\rightarrow$ `('A','B'), ('A','C'), ('B','C')` |
+| `itertools.islice(iterable, n)`| Schneidet die ersten $n$ Elemente aus einem Stream | `islice(count(), 5)` $\rightarrow$ `0, 1, 2, 3, 4` |
+
+---
+
+## 📁 Die Dateien in diesem Ordner
+
+- **`index.html`**: Interaktive Web-Lernseite mit visuellen Badewannen-Diagrammen, Code-Vergleichen, Cheat-Sheets und Checkliste.
+- **`aufgabe.py`**: Dein Arbeitsblatt mit Type Hints und didaktischen TODOs für Generatoren, Filter und `itertools`.
+- **`test_aufgabe.py`**: Automatische Unittest-Suite zur Überprüfung aller Anforderungen.
+- **`musterloesung.py`**: Vollständig ausprogrammierte Referenzlösung.
+
+---
+
+## 🚀 Schnellstart
+
+```bash
+# 1. Bearbeite aufgabe.py
+# 2. Führe die automatischen Tests aus:
+python3 test_aufgabe.py
+
+# 3. Oder starte das interaktive Benchmark-Skript:
+python3 aufgabe.py
+```
diff --git a/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/aufgabe.py b/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/aufgabe.py
new file mode 100644
index 0000000..f9963f3
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/aufgabe.py
@@ -0,0 +1,271 @@
+"""
+Kapitel 19: Generatoren, Iteratoren & itertools (Memory Efficiency & Streaming) 🌊⚡
+==================================================================================
+Lerne speichereffiziente Datenverarbeitung in Python:
+1. Iterables vs. Iterators (iter(), next(), StopIteration)
+2. Eigene Generatoren mit 'yield' (Werte on-demand generieren, 0 RAM-Verschwendung)
+3. Generator-Expressions (x**2 for x in riesige_daten)
+4. Das itertools-Modul (chain, cycle, count, combinations, permutations)
+
+Didaktische Analogie:
+- "Gefüllte Badewanne" (Listen): Alle Daten liegen gleichzeitig im RAM.
+- "Wasserhahn" (Generatoren): Daten fließen Tropfen für Tropfen, genau dann wenn sie gebraucht werden.
+"""
+
+import sys
+import itertools
+from typing import Iterator, Iterable, Any, Union, List, Tuple
+
+
+# ==============================================================================
+# TEIL 1: GRUNDLAGEN VON ITERATOREN & GENERATOREN MIT YIELD
+# ==============================================================================
+
+def endlos_zaehler(start: int = 0, schritt: int = 1) -> Iterator[int]:
+ """
+ Ein unendlicher Generator, der Zahlen beginnend bei 'start' mit der
+ Schrittweite 'schritt' generiert.
+
+ Parameter:
+ start (int): Die Startzahl (Standard: 0).
+ schritt (int): Die Schrittweite (Standard: 1).
+
+ Erzeugt (yield):
+ int: Die nächste Zahl in der unendlichen Sequenz.
+
+ Beispiele:
+ z = endlos_zaehler(10, 2)
+ next(z) -> 10
+ next(z) -> 12
+ next(z) -> 14
+ """
+ # ==========================================================================
+ # 🎯 TEILZIEL 1 (TODO 1): Implementiere endlos_zaehler
+ # 1. Initialisiere eine Variable 'aktuell' mit dem Wert 'start'.
+ # 2. Nutze eine unendliche Schleife (while True:).
+ # 3. Gib 'aktuell' mit 'yield' an den Aufrufer zurück.
+ # 4. Erhöhe 'aktuell' um 'schritt'.
+ # ==========================================================================
+ pass
+
+
+def fibonacci_generator(anzahl: int) -> Iterator[int]:
+ """
+ Ein Generator, der die ersten 'anzahl' Fibonacci-Zahlen on-demand erzeugt.
+ Die Fibonacci-Folge beginnt mit: 0, 1, 1, 2, 3, 5, 8, 13, 21, 34, ...
+
+ Parameter:
+ anzahl (int): Wie viele Fibonacci-Zahlen erzeugt werden sollen (>= 0).
+
+ Erzeugt (yield):
+ int: Die jeweils nächste Fibonacci-Zahl.
+
+ Exceptions:
+ ValueError: Falls 'anzahl' negativ (< 0) ist.
+
+ Beispiele:
+ list(fibonacci_generator(5)) -> [0, 1, 1, 2, 3]
+ list(fibonacci_generator(0)) -> []
+ """
+ # ==========================================================================
+ # 🎯 TEILZIEL 2 (TODO 2): Implementiere fibonacci_generator
+ # 1. Validiere 'anzahl': Wenn anzahl < 0, löse ValueError aus.
+ # 2. Setze die Startwerte a = 0, b = 1.
+ # 3. Wiederhole in einer Schleife 'anzahl'-mal:
+ # - Gib 'a' mit 'yield' zurück.
+ # - Aktualisiere die Werte mit Tupel-Zuweisung: a, b = b, a + b
+ # ==========================================================================
+ pass
+
+
+# ==============================================================================
+# TEIL 2: SPEICHERSCHONENDE FILTER & GENERATOR-EXPRESSIONS
+# ==============================================================================
+
+def filter_grosse_zahlen(zahlen_iterable: Iterable[Union[int, float]], schwellenwert: float) -> Iterator[Union[int, float]]:
+ """
+ Ein Generator-Filter, der nur Zahlen liefert, die strikt größer als
+ der angegebene 'schwellenwert' sind (x > schwellenwert).
+
+ Wichtig: Es darf KEINE Zwischenliste im Speicher aufgebaut werden!
+ Die Werte müssen einzeln gestreamt werden.
+
+ Parameter:
+ zahlen_iterable (Iterable): Eine beliebige Sequenz oder ein Generator von Zahlen.
+ schwellenwert (float): Der Grenzwert.
+
+ Erzeugt (yield):
+ Union[int, float]: Jede Zahl aus zahlen_iterable, die > schwellenwert ist.
+
+ Beispiele:
+ daten = [10, 5, 20, 3, 50]
+ list(filter_grosse_zahlen(daten, 15)) -> [20, 50]
+ """
+ # ==========================================================================
+ # 🎯 TEILZIEL 3 (TODO 3): Implementiere filter_grosse_zahlen
+ # 1. Iteriere mit einer for-Schleife über jedes Element in 'zahlen_iterable'.
+ # 2. Prüfe, ob das Element > schwellenwert ist.
+ # 3. Wenn ja, gib es mit 'yield' weiter.
+ # ==========================================================================
+ pass
+
+
+def erzeuge_quadrat_generator(n: int) -> Iterator[int]:
+ """
+ Erzeugt eine Generator-Expression für Quadratzahlen von 1 bis n (inklusive).
+
+ Parameter:
+ n (int): Die Obergrenze (>= 0).
+
+ Rückgabe:
+ Iterator[int]: Ein Generator-Objekt (x**2 for x in ...), keine Liste!
+
+ Exceptions:
+ ValueError: Falls n < 0 ist.
+
+ Beispiele:
+ gen = erzeuge_quadrat_generator(4)
+ list(gen) -> [1, 4, 9, 16]
+ """
+ # ==========================================================================
+ # 🎯 TEILZIEL 4 (TODO 4): Implementiere erzeuge_quadrat_generator
+ # 1. Wenn n < 0, löse einen ValueError aus.
+ # 2. Gib direkt eine Generator-Expression zurück: (x**2 for x in range(1, n + 1))
+ # ==========================================================================
+ pass
+
+
+# ==============================================================================
+# TEIL 3: DAS ITERTOOLS-MODUL (CHAIN, PERMUTATIONS & CYCLE)
+# ==============================================================================
+
+def kette_listen(*listen: Iterable[Any]) -> List[Any]:
+ """
+ Verbindet beliebig viele Iterables (*args) mit 'itertools.chain' zu einer
+ einzigen flachen Liste.
+
+ Parameter:
+ *listen (Iterable[Any]): Beliebig viele Iterables (Listen, Tupel, Generatoren).
+
+ Rückgabe:
+ List[Any]: Eine zusammenhängende Liste aller Elemente.
+
+ Beispiele:
+ kette_listen([1, 2], [3, 4], [5]) -> [1, 2, 3, 4, 5]
+ kette_listen([], [10]) -> [10]
+ """
+ # ==========================================================================
+ # 🎯 TEILZIEL 5 (TODO 5): Implementiere kette_listen mit itertools.chain
+ # 1. Nutze itertools.chain(*listen) oder itertools.chain.from_iterable(listen).
+ # 2. Wandle das Ergebnis in eine normale Python-Liste um (list(...)) und gib sie zurück.
+ # ==========================================================================
+ pass
+
+
+def erzeuge_passwort_kombinationen(zeichen: str, laenge: int) -> List[str]:
+ """
+ Erzeugt alle möglichen Permutationen der angegebenen Zeichenkette mit der
+ exakten Länge 'laenge' unter Verwendung von 'itertools.permutations'.
+
+ Parameter:
+ zeichen (str): Die verfügbaren Zeichen als String (z.B. "abc").
+ laenge (int): Die Länge der gesuchten Kombinationen.
+
+ Rückgabe:
+ List[str]: Eine Liste von Strings mit allen Permutationen.
+
+ Exceptions:
+ ValueError: Falls laenge < 0 oder laenge > len(zeichen) ist.
+
+ Beispiele:
+ erzeuge_passwort_kombinationen("abc", 2) -> ['ab', 'ac', 'ba', 'bc', 'ca', 'cb']
+ erzeuge_passwort_kombinationen("ab", 1) -> ['a', 'b']
+ """
+ # ==========================================================================
+ # 🎯 TEILZIEL 6 (TODO 6): Implementiere erzeuge_passwort_kombinationen
+ # 1. Validiere: Wenn laenge < 0 oder laenge > len(zeichen), löse ValueError aus.
+ # 2. Verwende itertools.permutations(zeichen, laenge).
+ # 3. Führe jedes generierte Tupel von Zeichen mit "".join(...) zu einem String zusammen.
+ # 4. Gib alle kombinierten Strings als Liste zurück.
+ # ==========================================================================
+ pass
+
+
+def zyklische_elemente(elemente: List[Any], anzahl: int) -> List[Any]:
+ """
+ Nimmt eine Liste von Elementen und liefert die ersten 'anzahl' Elemente
+ in zyklischer Reihenfolge mit 'itertools.cycle'.
+
+ Parameter:
+ elemente (List[Any]): Die Ausgangsliste (darf nicht leer sein, falls anzahl > 0).
+ anzahl (int): Wie viele Elemente zyklisch entnommen werden sollen.
+
+ Rückgabe:
+ List[Any]: Liste mit 'anzahl' Elementen.
+
+ Exceptions:
+ ValueError: Falls anzahl < 0 ist oder falls elemente leer ist bei anzahl > 0.
+
+ Beispiele:
+ zyklische_elemente(["Rot", "Gruen", "Blau"], 7)
+ -> ["Rot", "Gruen", "Blau", "Rot", "Gruen", "Blau", "Rot"]
+ """
+ # ==========================================================================
+ # 🎯 TEILZIEL 7 (TODO 7): Implementiere zyklische_elemente
+ # 1. Validiere anzahl >= 0 (sonst ValueError).
+ # 2. Wenn anzahl == 0, gib eine leere Liste [] zurück.
+ # 3. Wenn elemente leer ist (und anzahl > 0), löse ValueError aus.
+ # 4. Erstelle einen zyklischen Iterator mit zyk = itertools.cycle(elemente).
+ # 5. Hole mit next(zyk) genau 'anzahl' Elemente ab und sammle sie in einer Liste.
+ # ==========================================================================
+ pass
+
+
+# ==============================================================================
+# DEMONSTRATION & SPEICHER-BENCHMARK
+# ==============================================================================
+
+if __name__ == "__main__":
+ print("=" * 70)
+ print("🌊 KAPITEL 19: GENERATOREN & ITERTOOLS (MEMORY EFFICIENCY)")
+ print("=" * 70)
+
+ # 1. Speichervergleich: Liste vs. Generator
+ elemente_anzahl = 1_000_000
+ print(f"\n📊 SPEICHERVERGLEICH FÜR {elemente_anzahl:,} ZAHLEN:")
+
+ liste = [x for x in range(elemente_anzahl)]
+ gen = (x for x in range(elemente_anzahl))
+
+ speicher_liste = sys.getsizeof(liste) / (1024 * 1024)
+ speicher_gen = sys.getsizeof(gen)
+
+ print(f" 🔴 Liste (Badewanne): {speicher_liste:.2f} MB RAM im Hauptspeicher!")
+ print(f" 🟢 Generator (Wasserhahn): {speicher_gen} Bytes RAM (Faktor {int(speicher_liste * 1024 * 1024 / speicher_gen):,}x sparsamer!)")
+
+ # 2. Endloszähler Test
+ print("\n⏱️ Endloszähler (erste 5 Werte ab 100 in 10er Schritten):")
+ zaehler = endlos_zaehler(100, 10)
+ if zaehler:
+ for _ in range(5):
+ print(f" -> {next(zaehler)}")
+
+ # 3. Fibonacci Test
+ print("\n🐰 Die ersten 10 Fibonacci-Zahlen:")
+ fibs = fibonacci_generator(10)
+ if fibs:
+ print(f" -> {list(fibs)}")
+
+ # 4. Passwort-Permutationen Test
+ print("\n🔐 Passwort-Permutationen für 'abc' (Länge 2):")
+ perms = erzeuge_passwort_kombinationen("abc", 2)
+ print(f" -> {perms}")
+
+ # 5. Zyklischer Taktgeber
+ print("\n🚦 Ampel-Zyklus (7 Schritte):")
+ ampel = zyklische_elemente(["Rot", "Gelb", "Gruen"], 7)
+ print(f" -> {ampel}")
+
+ print("\n" + "=" * 70)
+ print("💡 Starte die Tests mit: python3 test_aufgabe.py")
+ print("=" * 70)
diff --git a/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/index.html b/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/index.html
new file mode 100644
index 0000000..1650db3
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/19_generatoren_und_itertools/index.html
@@ -0,0 +1,502 @@
+
+
+
+
+
+ Kapitel 19: Generatoren, Iteratoren & itertools
+
+
+
+
+
+
+
+
+
+
+
+
Lehrpfad 4 • Professional Data Engineering
+
Generatoren, Iteratoren & itertools 🌊⚡
+
Verarbeite riesige Datenströme mit 0 RAM-Verschwendung: Entdecke das Geheimnis von yield, Lazy Evaluation und Pythons High-Performance-Modul itertools.
3. Magie im Einsatz: Eigene Generatoren mit yield 🪄
+
+
+ Im Gegensatz zu return (welches eine Funktion komplett beendet) unterbricht yield die Funktion nur kurzzeitig, gibt den Wert zurück und friert den gesamten Zustand (lokale Variablen, Zeilennummer) ein!
+
+
+
+
Endloszähler & Fibonacci mit yield
+
defendlos_zaehler(start=0, schritt=1):
+ aktuell = start
+ whileTrue:
+ yieldaktuell# Pausiert hier und gibt den Wert zurück
+ aktuell += schritt# Macht beim nächsten next() genau hier weiter!
+
+z = endlos_zaehler(10, 5)
+print(next(z)) # -> 10
+print(next(z)) # -> 15
+print(next(z)) # -> 20
+
+
+
+
+
+
4. Das itertools Power-Modul 🧰
+
+
+ Das eingebaute Modul itertools bietet blitzschnelle, in C geschriebene Iterator-Bausteine:
+
+
+
+
+
+
Werkzeug
+
Zweck
+
Beispiel
+
+
+
+
+
itertools.chain(*iterables)
+
Verkettet mehrere Listen/Generatoren ohne Kopieren
from dataclasses import dataclass, field
+
+@dataclass
+classArtikel:
+ name: str
+ preis: float
+ kategorie: str = "Allgemein"
+
+ ✨ __init__, __repr__ und __eq__ sind automatisch fertig!
+
+
+
+
+
⚠️ Wichtig: Veränderliche Default-Listen mit field(default_factory=list)
+
Schreibe in einer Dataclass niemals liste: list = [], da sonst alle Instanzen dieselbe Liste teilen! Nutze stattdessen immer field(default_factory=list).
+
+
+
+
+
+
2. Moderne Type Hints im Überblick 🏷️
+
+
+ Type Hints machen deinen Code selbsterklärend und ermöglichen IDEs wie VS Code blitzschnelle Autovervollständigung und Fehlerwarnungen:
+
+
+
+
+
+
Type Hint
+
Bedeutung
+
Praxis-Beispiel
+
+
+
+
+
Optional[str]
+
Kann str oder None sein
+
def suche(id: int) -> Optional[User]: ...
+
+
+
Union[int, float]
+
Zahl kann Ganzzahl oder Fließkomma sein
+
def runden(wert: Union[int, float]) -> float: ...
+
+
+
Callable[[A, B], C]
+
Eine übergebene Funktion
+
def ausfuehren(func: Callable) -> None: ...
+
+
+
Dict[str, Any]
+
Schlüssel sind Strings, Werte beliebig
+
user_config: Dict[str, Any] = {"theme": "dark"}
+
+
+
+
+
+
+
+
3. Eigene Decorators schreiben: Das Geschenkverpackungs-Muster 🎁
+
+
+ Ein Decorator ist eine Funktion, die eine andere Funktion einpackt und ihr zusätzliche Fähigkeiten verleiht (z.B. Zeitmessung, Logging, Aufrufzählung):
+
+
+
+
Eigener Zeitmessungs-Decorator mit @functools.wraps
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/lehrpfad_4_professional_data_engineering/20_dataclasses_und_typehints/musterloesung.py b/lehrpfad_4_professional_data_engineering/20_dataclasses_und_typehints/musterloesung.py
new file mode 100644
index 0000000..4eea2ff
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/20_dataclasses_und_typehints/musterloesung.py
@@ -0,0 +1,212 @@
+"""
+Kapitel 20: Dataclasses, Type Hints & Decorators (Moderne Python-Architektur) 🏷️✨
+==================================================================================
+Musterlösung für moderne Python-Architektur mit Dataclasses, Typisierung und Decorators.
+"""
+
+import time
+import functools
+from dataclasses import dataclass, field
+from typing import Optional, Union, Callable, List, Dict, Any
+
+
+# ==============================================================================
+# TEIL 1: DATACLASSES FÜR TYPSICHERE DATENMODELLE
+# ==============================================================================
+
+@dataclass
+class Artikel:
+ """
+ Ein Artikel im Online-Shop als moderne Dataclass.
+ """
+ name: str
+ preis: float
+ kategorie: str = "Allgemein"
+
+ def __post_init__(self) -> None:
+ if not self.name or not str(self.name).strip():
+ raise ValueError("Artikelname darf nicht leer sein!")
+ if float(self.preis) < 0.0:
+ raise ValueError("Preis darf nicht negativ sein!")
+
+ def berechne_bruttopreis(self, mwst_satz: float = 0.19) -> float:
+ """
+ Berechnet den Bruttopreis inklusive Mehrwertsteuer, gerundet auf 2 Nachkommastellen.
+ """
+ if float(mwst_satz) < 0.0:
+ raise ValueError("Mehrwertsteuersatz darf nicht negativ sein!")
+ return round(float(self.preis) * (1.0 + float(mwst_satz)), 2)
+
+ @classmethod
+ def aus_csv_zeile(cls, csv_zeile: str) -> "Artikel":
+ """
+ Erstellt ein Artikel-Objekt aus einer CSV-Zeile wie 'Laptop;999.99;Elektronik'.
+ """
+ teile = [t.strip() for t in csv_zeile.split(";")]
+ if len(teile) < 2:
+ raise ValueError("CSV-Zeile muss mindestens Name und Preis enthalten!")
+
+ name = teile[0]
+ preis = float(teile[1])
+ kategorie = teile[2] if len(teile) > 2 and teile[2] else "Allgemein"
+ return cls(name=name, preis=preis, kategorie=kategorie)
+
+ @staticmethod
+ def ist_gueltiger_preis(preis: Any) -> bool:
+ """
+ Prüft statisch, ob ein Preis gültig (numerisch und >= 0) ist.
+ """
+ if isinstance(preis, (int, float)) and not isinstance(preis, bool):
+ return preis >= 0.0
+ return False
+
+
+@dataclass
+class Warenkorb:
+ """
+ Ein digitaler Warenkorb, der Artikel sammelt und verwaltet.
+ """
+ kunde: str
+ artikel_liste: List[Artikel] = field(default_factory=list)
+
+ def artikel_hinzufuegen(self, *artikel: Artikel) -> None:
+ """
+ Fügt beliebig viele Artikel (*args) zum Warenkorb hinzu.
+ """
+ for a in artikel:
+ if not isinstance(a, Artikel):
+ raise TypeError(f"Erwartete Artikel-Instanz, erhielt: {type(a).__name__}")
+ self.artikel_liste.append(a)
+
+ def artikel_entfernen(self, artikel_name: str) -> bool:
+ """
+ Entfernt den ersten Artikel mit dem passenden Namen (Groß-/Kleinschreibung ignorieren).
+ """
+ such_name = artikel_name.strip().lower()
+ for a in self.artikel_liste:
+ if a.name.strip().lower() == such_name:
+ self.artikel_liste.remove(a)
+ return True
+ return False
+
+ def gesamtsumme(self, brutto: bool = False, mwst_satz: float = 0.19) -> float:
+ """
+ Berechnet die Gesamtsumme aller Artikel im Warenkorb.
+ """
+ if not self.artikel_liste:
+ return 0.0
+
+ if brutto:
+ summe = sum(a.berechne_bruttopreis(mwst_satz) for a in self.artikel_liste)
+ else:
+ summe = sum(a.preis for a in self.artikel_liste)
+
+ return round(summe, 2)
+
+ def __len__(self) -> int:
+ return len(self.artikel_liste)
+
+
+# ==============================================================================
+# TEIL 2: EIGENE DECORATORS SCHREIBEN
+# ==============================================================================
+
+def aufruf_zaehler(func: Callable) -> Callable:
+ """
+ Ein Decorator, der zählt, wie oft eine dekorierte Funktion aufgerufen wurde.
+ """
+ @functools.wraps(func)
+ def wrapper(*args: Any, **kwargs: Any) -> Any:
+ wrapper.aufrufe += 1
+ return func(*args, **kwargs)
+
+ wrapper.aufrufe = 0
+ return wrapper
+
+
+def zeitmessung(func: Callable) -> Callable:
+ """
+ Ein Decorator, der die Ausführungszeit einer Funktion misst.
+ """
+ @functools.wraps(func)
+ def wrapper(*args: Any, **kwargs: Any) -> Any:
+ start = time.perf_counter()
+ ergebnis = func(*args, **kwargs)
+ dauer = time.perf_counter() - start
+ wrapper.letzte_dauer = dauer
+ return ergebnis
+
+ wrapper.letzte_dauer = 0.0
+ return wrapper
+
+
+# ==============================================================================
+# TEIL 3: FORTGESCHRITTENE TYPE HINTS (UNION, DICT, TYPED DATA)
+# ==============================================================================
+
+def formatiere_benutzer(daten: Dict[str, Union[str, int, bool]]) -> str:
+ """
+ Formatiert ein Benutzer-Dictionary mit verschiedenen Datentypen zu einem
+ lesbaren Status-String.
+ """
+ if "username" not in daten and "name" not in daten:
+ raise KeyError("Benutzerdaten müssen 'username' oder 'name' enthalten!")
+
+ username_val = daten.get("username", daten.get("name"))
+ if not username_val or not str(username_val).strip():
+ raise ValueError("Benutzername darf nicht leer sein!")
+
+ username = str(username_val).strip()
+
+ alter_val = daten.get("alter")
+ alter_str = str(alter_val) if alter_val is not None else "Unbekannt"
+
+ ist_admin = daten.get("ist_admin", False)
+ rolle = "Administrator" if ist_admin else "Standardbenutzer"
+
+ return f"Benutzer: {username} | Alter: {alter_str} | Rolle: {rolle}"
+
+
+# ==============================================================================
+# DEMONSTRATION & MAIN
+# ==============================================================================
+
+if __name__ == "__main__":
+ print("=" * 70)
+ print("🏷️ KAPITEL 20: MUSTERLÖSUNG – DATACLASSES & DECORATORS")
+ print("=" * 70)
+
+ # 1. Dataclass Artikel testen
+ laptop = Artikel(name="ThinkPad X1", preis=1499.00, kategorie="Elektronik")
+ print(f"\n📦 Artikel Dataclass:\n {laptop}")
+ print(f" Bruttopreis (19%): {laptop.berechne_bruttopreis(0.19):.2f} €")
+
+ # 2. CSV Fabrikmethode
+ maus = Artikel.aus_csv_zeile("Logitech MX Master;99.90;Zubehoer")
+ print(f" Aus CSV erstellt: {maus}")
+
+ # 3. Warenkorb testen
+ korb = Warenkorb(kunde="Max Mustermann")
+ korb.artikel_hinzufuegen(laptop, maus, Artikel("Kaffeebecher", 12.50, "Haushalt"))
+ print(f"\n🛒 Warenkorb für {korb.kunde} ({len(korb)} Artikel):")
+ print(f" Nettosumme: {korb.gesamtsumme():.2f} €")
+ print(f" Bruttosumme: {korb.gesamtsumme(brutto=True):.2f} €")
+
+ # 4. Aufruf-Zähler Decorator
+ @aufruf_zaehler
+ def sage_hallo(name: str) -> str:
+ return f"Hallo, {name}!"
+
+ print("\n🔢 Aufruf-Zähler Decorator Test:")
+ print(sage_hallo("Alice"))
+ print(sage_hallo("Bob"))
+ print(sage_hallo("Charlie"))
+ print(f" Anzahl Aufrufe: {sage_hallo.aufrufe}")
+
+ # 5. Typsicherer Benutzer-Formatierer
+ user = {"username": "CoderGirl99", "alter": 24, "ist_admin": True}
+ print(f"\n👤 Benutzerdaten:\n {formatiere_benutzer(user)}")
+
+ print("\n" + "=" * 70)
+ print("✅ Musterlösung erfolgreich ausgeführt!")
+ print("=" * 70)
diff --git a/lehrpfad_4_professional_data_engineering/20_dataclasses_und_typehints/test_aufgabe.py b/lehrpfad_4_professional_data_engineering/20_dataclasses_und_typehints/test_aufgabe.py
new file mode 100644
index 0000000..f93e904
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/20_dataclasses_und_typehints/test_aufgabe.py
@@ -0,0 +1,191 @@
+import unittest
+import sys
+import time
+from pathlib import Path
+from dataclasses import is_dataclass
+
+# Sicherstellen, dass das aktuelle Verzeichnis im Python-Pfad liegt
+sys.path.insert(0, str(Path(__file__).parent))
+
+try:
+ import aufgabe
+except ImportError as e:
+ raise ImportError(f"Konnte 'aufgabe.py' nicht importieren: {e}")
+
+
+class TestKapitel20(unittest.TestCase):
+ """
+ Testsuite für Kapitel 20: Dataclasses, Type Hints & Decorators.
+ """
+
+ def setUp(self):
+ self.Artikel = getattr(aufgabe, "Artikel", None)
+ self.Warenkorb = getattr(aufgabe, "Warenkorb", None)
+ self.aufruf_zaehler = getattr(aufgabe, "aufruf_zaehler", None)
+ self.zeitmessung = getattr(aufgabe, "zeitmessung", None)
+ self.formatiere_benutzer = getattr(aufgabe, "formatiere_benutzer", None)
+
+ self.assertIsNotNone(self.Artikel, "Klasse 'Artikel' nicht gefunden!")
+ self.assertIsNotNone(self.Warenkorb, "Klasse 'Warenkorb' nicht gefunden!")
+ self.assertIsNotNone(self.aufruf_zaehler, "Decorator 'aufruf_zaehler' nicht gefunden!")
+ self.assertIsNotNone(self.zeitmessung, "Decorator 'zeitmessung' nicht gefunden!")
+ self.assertIsNotNone(self.formatiere_benutzer, "Funktion 'formatiere_benutzer' nicht gefunden!")
+
+ # ==========================================================================
+ # 1. Dataclass Artikel
+ # ==========================================================================
+ def test_01_artikel_dataclass_basis(self):
+ """Prüft, ob Artikel eine echte Dataclass ist und Standardwerte setzt."""
+ self.assertTrue(is_dataclass(self.Artikel), "Artikel muss mit @dataclass dekoriert sein!")
+
+ a = self.Artikel("Buch", 19.99)
+ self.assertEqual(a.name, "Buch")
+ self.assertEqual(a.preis, 19.99)
+ self.assertEqual(a.kategorie, "Allgemein")
+
+ # Gleichheit und Repräsentation (Dataclass-Features)
+ a2 = self.Artikel("Buch", 19.99)
+ self.assertEqual(a, a2, "Zwei Artikel mit identischen Werten müssen gleich (==) sein!")
+ self.assertIn("Buch", repr(a))
+
+ def test_02_artikel_validierung(self):
+ """Prüft die Validierung in __post_init__."""
+ with self.assertRaises(ValueError, msg="Negativer Preis muss ValueError auslösen!"):
+ self.Artikel("Test", -5.0)
+
+ with self.assertRaises(ValueError, msg="Leerer Name muss ValueError auslösen!"):
+ self.Artikel("", 10.0)
+
+ with self.assertRaises(ValueError, msg="Nur-Leerzeichen-Name muss ValueError auslösen!"):
+ self.Artikel(" ", 10.0)
+
+ def test_03_artikel_berechne_bruttopreis(self):
+ """Prüft die Bruttoberechnung mit verschiedenen Steuersätzen."""
+ buch = self.Artikel("Buch", 10.0)
+ self.assertEqual(buch.berechne_bruttopreis(0.07), 10.70)
+ self.assertEqual(buch.berechne_bruttopreis(0.19), 11.90)
+ self.assertEqual(buch.berechne_bruttopreis(0.0), 10.0)
+
+ with self.assertRaises(ValueError, msg="Negativer Steuersatz muss ValueError werfen"):
+ buch.berechne_bruttopreis(-0.1)
+
+ def test_04_artikel_classmethod_und_staticmethod(self):
+ """Prüft @classmethod aus_csv_zeile und @staticmethod ist_gueltiger_preis."""
+ # Classmethod
+ a = self.Artikel.aus_csv_zeile("Monitor;249.99;Hardware")
+ self.assertEqual(a.name, "Monitor")
+ self.assertEqual(a.preis, 249.99)
+ self.assertEqual(a.kategorie, "Hardware")
+
+ # Staticmethod
+ self.assertTrue(self.Artikel.ist_gueltiger_preis(100))
+ self.assertTrue(self.Artikel.ist_gueltiger_preis(0.0))
+ self.assertFalse(self.Artikel.ist_gueltiger_preis(-10))
+ self.assertFalse(self.Artikel.ist_gueltiger_preis("100"))
+
+ # ==========================================================================
+ # 2. Dataclass Warenkorb
+ # ==========================================================================
+ def test_05_warenkorb_basis_und_default_factory(self):
+ """Prüft Warenkorb, default_factory und Unabhängigkeit von Instanzen."""
+ self.assertTrue(is_dataclass(self.Warenkorb), "Warenkorb muss mit @dataclass dekoriert sein!")
+
+ k1 = self.Warenkorb(kunde="Alice")
+ k2 = self.Warenkorb(kunde="Bob")
+
+ self.assertEqual(len(k1), 0)
+ self.assertEqual(k1.gesamtsumme(), 0.0)
+
+ # Sicherstellen, dass default_factory=list genutzt wurde (keine geteilte Referenz)
+ k1.artikel_hinzufuegen(self.Artikel("Tee", 3.50))
+ self.assertEqual(len(k1), 1)
+ self.assertEqual(len(k2), 0, "artikel_liste darf nicht zwischen Warenkörben geteilt werden!")
+
+ def test_06_warenkorb_artikel_hinzufuegen_und_entfernen(self):
+ """Prüft Hinzufügen (*args), Typprüfung und Entfernen von Artikeln."""
+ korb = self.Warenkorb(kunde="Charlie")
+ a1 = self.Artikel("Kaffee", 6.00)
+ a2 = self.Artikel("Keks", 2.00)
+
+ korb.artikel_hinzufuegen(a1, a2)
+ self.assertEqual(len(korb), 2)
+ self.assertEqual(korb.gesamtsumme(), 8.00)
+ self.assertEqual(korb.gesamtsumme(brutto=True, mwst_satz=0.10), 8.80)
+
+ # Typprüfung
+ with self.assertRaises(TypeError, msg="Ungültige Objekte müssen TypeError auslösen!"):
+ korb.artikel_hinzufuegen("Kein Artikel")
+
+ # Artikel entfernen
+ entfernt = korb.artikel_entfernen("kaffee") # Case-insensitive
+ self.assertTrue(entfernt)
+ self.assertEqual(len(korb), 1)
+ self.assertEqual(korb.gesamtsumme(), 2.00)
+
+ # Nicht vorhandener Artikel
+ self.assertFalse(korb.artikel_entfernen("NichtDa"))
+
+ # ==========================================================================
+ # 3. Decorator aufruf_zaehler
+ # ==========================================================================
+ def test_07_aufruf_zaehler_decorator(self):
+ """Prüft, ob der Decorator Aufrufe zählt und functools.wraps verwendet."""
+ @self.aufruf_zaehler
+ def addiere(a: int, b: int) -> int:
+ """Addiert zwei Zahlen."""
+ return a + b
+
+ self.assertEqual(addiere.__name__, "addiere", "functools.wraps muss den Funktionsnamen bewahren!")
+ self.assertEqual(addiere.__doc__, "Addiert zwei Zahlen.", "functools.wraps muss den Docstring bewahren!")
+ self.assertEqual(addiere.aufrufe, 0)
+
+ res1 = addiere(5, 10)
+ self.assertEqual(res1, 15)
+ self.assertEqual(addiere.aufrufe, 1)
+
+ res2 = addiere(20, 30)
+ self.assertEqual(res2, 50)
+ self.assertEqual(addiere.aufrufe, 2)
+
+ # ==========================================================================
+ # 4. Decorator zeitmessung
+ # ==========================================================================
+ def test_08_zeitmessung_decorator(self):
+ """Prüft, ob zeitmessung die Dauer erfasst und das Ergebnis zurückgibt."""
+ @self.zeitmessung
+ def schlafe_kurz():
+ time.sleep(0.01)
+ return "Aufgewacht"
+
+ ergebnis = schlafe_kurz()
+ self.assertEqual(ergebnis, "Aufgewacht")
+ self.assertTrue(hasattr(schlafe_kurz, "letzte_dauer"))
+ self.assertGreaterEqual(schlafe_kurz.letzte_dauer, 0.005)
+
+ # ==========================================================================
+ # 5. Type Hints formatiere_benutzer
+ # ==========================================================================
+ def test_09_formatiere_benutzer(self):
+ """Prüft die Formatierung typisierter Benutzer-Dictionaries und Fehler."""
+ u1 = {"username": "Alice", "alter": 30, "ist_admin": True}
+ res1 = self.formatiere_benutzer(u1)
+ self.assertIn("Alice", res1)
+ self.assertIn("30", res1)
+ self.assertIn("Administrator", res1)
+
+ u2 = {"username": "Bob", "ist_admin": False}
+ res2 = self.formatiere_benutzer(u2)
+ self.assertIn("Bob", res2)
+ self.assertIn("Unbekannt", res2)
+ self.assertIn("Standardbenutzer", res2)
+
+ # Fehler bei fehlendem Username
+ with self.assertRaises(KeyError):
+ self.formatiere_benutzer({"alter": 20})
+
+ with self.assertRaises(ValueError):
+ self.formatiere_benutzer({"username": " "})
+
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)
diff --git a/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/README.md b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/README.md
new file mode 100644
index 0000000..c440123
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/README.md
@@ -0,0 +1,48 @@
+# Kapitel 21: Relationale Datenbanken & SQLite (sqlite3) 🗄️⚡
+
+In diesem Kapitel lernst du, wie du strukturierte relationale Datenbanken mit Pythons eingebautem Modul `sqlite3` erstellst, verwaltest und sicher abfragst.
+
+---
+
+## 🎯 Was du lernst
+
+1. **Relationale Datenbanken vs. Einfache Textdateien:** Warum relationale Datenbanken (Tabellen, Primärschlüssel, Relationen) für große Datenmengen und gleichzeitigen Zugriff unverzichtbar sind.
+2. **Die Aktenschrank-Analogie:**
+ - **Datenbankdatei:** Der Aktenschrank.
+ - **Tabelle:** Eine Schublade mit Registerkarten.
+ - **Spalte (Column):** Ein vorgegebenes Feld auf der Registerkarte (z.B. Name, E-Mail, Guthaben).
+ - **Zeile (Row/Record):** Eine einzelne ausgefüllte Registerkarte (ein Kunde).
+3. **Verbindung & Cursor:**
+ ```python
+ import sqlite3
+
+ conn = sqlite3.connect("kunden.db") # oder ":memory:" für flüchtige RAM-Datenbank
+ conn.row_factory = sqlite3.Row # Zugriff per Spaltennamen wie im Dict!
+ cursor = conn.cursor()
+ ```
+4. **CRUD-Operationen in SQL:**
+ - **C**reate: `CREATE TABLE IF NOT EXISTS ...` und `INSERT INTO ...`
+ - **R**ead: `SELECT ... FROM ... WHERE ...`
+ - **U**pdate: `UPDATE ... SET ... WHERE ...`
+ - **D**elete: `DELETE FROM ... WHERE ...`
+5. **Transaktionen & Persistenz:** Warum Änderungen erst nach `conn.commit()` dauerhaft auf der Festplatte gespeichert werden.
+6. **Sicherheit & SQL-Injection-Schutz:** Niemals Strings formatieren (`f"SELECT ... {user_input}"`), sondern immer Platzhalter (`?`) verwenden!
+
+---
+
+## 📁 Die Dateien in diesem Ordner
+
+- **`index.html`**: Die interaktive Lernseite mit Aktenschrank-Analogie, SQL-Cheat-Sheet, Injection-Demo und Subgoals.
+- **`aufgabe.py`**: Dein Arbeitsblatt mit der Klasse `KundenDatenbank`.
+- **`test_aufgabe.py`**: Automatische Unittest-Suite (`python3 test_aufgabe.py`).
+- **`musterloesung.py`**: Vollständig ausprogrammierte Beispiellösung.
+
+---
+
+## 🚀 Schnellstart
+
+```bash
+# 1. Bearbeite aufgabe.py
+# 2. Teste deine Lösung:
+python3 test_aufgabe.py
+```
diff --git a/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/aufgabe.py b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/aufgabe.py
new file mode 100644
index 0000000..e88d599
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/aufgabe.py
@@ -0,0 +1,169 @@
+"""
+Kapitel 21: Relationale Datenbanken & SQLite (sqlite3) 🗄️⚡
+=========================================================
+Lehrpfad 4: Professional Data Engineering
+
+Thema:
+Relationales Datenmanagement mit SQLite und Pythons eingebautem Modul `sqlite3`.
+
+Lernziele:
+1. Verbindungsaufbau mit `sqlite3.connect()` und Arbeiten mit dem Cursor `conn.cursor()`.
+2. Tabellen definieren mit `CREATE TABLE IF NOT EXISTS`.
+3. Datensätze einfügen mit `INSERT INTO` und `conn.commit()`.
+4. Daten gezielt abfragen mit `SELECT ... WHERE` und `fetchall()` / `fetchone()`.
+5. Datensätze aktualisieren mit `UPDATE ... SET` und löschen mit `DELETE FROM`.
+6. Schutz vor SQL-Injections durch parametrisierte Abfragen (`?`).
+7. Verwendung von `row_factory = sqlite3.Row` für benannte Spaltenzugriffe.
+"""
+
+import sqlite3
+from typing import List, Optional, Dict, Any
+
+
+class KundenDatenbank:
+ """
+ Verwaltet Kundenkonten und Guthaben in einer relationalen SQLite-Datenbank.
+ """
+
+ def __init__(self, db_pfad: str = ":memory:"):
+ """
+ Initialisiert die Datenbankverbindung und erstellt die Tabelle 'kunden',
+ falls sie noch nicht existiert.
+
+ Tabelle 'kunden':
+ - id: INTEGER PRIMARY KEY AUTOINCREMENT
+ - name: TEXT NOT NULL
+ - email: TEXT NOT NULL UNIQUE
+ - guthaben: REAL DEFAULT 0.0
+
+ Tipps:
+ - self.db_pfad = db_pfad
+ - self.conn = sqlite3.connect(db_pfad)
+ - self.conn.row_factory = sqlite3.Row # Ermöglicht dict-artigen Zugriff
+ - self.cursor = self.conn.cursor()
+ - Führe den CREATE TABLE Befehl mit self.cursor.execute(...) aus
+ - Führe self.conn.commit() aus
+ """
+ # TODO: Implementieren
+ pass
+
+ def kunde_hinzufuegen(self, name: str, email: str, startguthaben: float = 0.0) -> int:
+ """
+ Fügt einen neuen Kunden in die Datenbank ein.
+
+ WICHTIG:
+ - Verwende IMMER parametrisierte SQL-Queries mit '?' gegen SQL-Injections!
+ - SQL: "INSERT INTO kunden (name, email, guthaben) VALUES (?, ?, ?)"
+ - Führe nach dem INSERT self.conn.commit() auf.
+
+ Rückgabewert:
+ - Die automatisch vergebene ID des neuen Kunden (int) via self.cursor.lastrowid
+ """
+ # TODO: Implementieren
+ pass
+
+ def kunde_suchen_nach_id(self, kunden_id: int) -> Optional[Dict[str, Any]]:
+ """
+ Sucht einen Kunden anhand seiner ID.
+
+ SQL:
+ "SELECT id, name, email, guthaben FROM kunden WHERE id = ?"
+
+ Rückgabewert:
+ - Ein Dictionary mit den Spalten als Keys:
+ {"id": row["id"], "name": row["name"], "email": row["email"], "guthaben": row["guthaben"]}
+ - None, falls kein Kunde mit dieser ID gefunden wurde.
+ """
+ # TODO: Implementieren
+ pass
+
+ def guthaben_aufladen(self, kunden_id: int, betrag: float) -> bool:
+ """
+ Lädt das Guthaben eines bestehenden Kunden um den angegebenen Betrag auf.
+
+ Bedingungen:
+ - betrag muss größer als 0 sein (andernfalls direkt False zurückgeben).
+ - SQL: "UPDATE kunden SET guthaben = guthaben + ? WHERE id = ?"
+ - Führe self.conn.commit() aus.
+
+ Rückgabewert:
+ - True, wenn der Datensatz aktualisiert wurde (self.cursor.rowcount > 0).
+ - False, wenn kein Kunde mit dieser ID existiert oder betrag <= 0 ist.
+ """
+ # TODO: Implementieren
+ pass
+
+ def kunden_mit_mindestguthaben(self, min_guthaben: float) -> List[Dict[str, Any]]:
+ """
+ Gibt alle Kunden zurück, deren Guthaben >= min_guthaben ist.
+ Sortiert nach Guthaben absteigend (höchstes Guthaben zuerst).
+
+ SQL:
+ "SELECT id, name, email, guthaben FROM kunden WHERE guthaben >= ? ORDER BY guthaben DESC"
+
+ Rückgabewert:
+ - Liste von Dictionaries der passenden Kunden.
+ [{"id": ..., "name": ..., "email": ..., "guthaben": ...}, ...]
+ """
+ # TODO: Implementieren
+ pass
+
+ def kunde_loeschen(self, kunden_id: int) -> bool:
+ """
+ Löscht einen Kunden anhand seiner ID aus der Datenbank.
+
+ SQL:
+ "DELETE FROM kunden WHERE id = ?"
+
+ Rückgabewert:
+ - True, wenn ein Kunde gelöscht wurde (self.cursor.rowcount > 0).
+ - False, wenn kein Kunde mit dieser ID existierte.
+ """
+ # TODO: Implementieren
+ pass
+
+ def alle_kunden(self) -> List[Dict[str, Any]]:
+ """
+ Gibt eine Liste aller Kunden zurück.
+ """
+ # TODO: Implementieren
+ pass
+
+ def schliessen(self) -> None:
+ """Schließt die Datenbankverbindung."""
+ if hasattr(self, "conn") and self.conn:
+ self.conn.close()
+
+
+# ==============================================================================
+# Interaktiver Test im Terminal:
+# (python3 aufgabe.py)
+# ==============================================================================
+if __name__ == "__main__":
+ print("🗄️ Initialisiere KundenDatenbank im Arbeitsspeicher (:memory:)...")
+ db = KundenDatenbank(":memory:")
+
+ # 1. Kunden anlegen
+ id1 = db.kunde_hinzufuegen("Alice Schmidt", "alice@example.com", 150.0)
+ id2 = db.kunde_hinzufuegen("Bob Müller", "bob@example.com", 50.0)
+ id3 = db.kunde_hinzufuegen("Charlie Braun", "charlie@example.com", 320.0)
+ print(f"Kunden erstellt: IDs = {id1}, {id2}, {id3}")
+
+ # 2. Kunde suchen
+ kunde = db.kunde_suchen_nach_id(id1)
+ print(f"Gefundener Kunde: {kunde}")
+
+ # 3. Guthaben aufladen
+ erfolg = db.guthaben_aufladen(id2, 100.0)
+ print(f"Guthaben von Bob aufgeladen: {erfolg} -> Neuer Stand: {db.kunde_suchen_nach_id(id2)}")
+
+ # 4. Mindestguthaben filtern
+ reiche_kunden = db.kunden_mit_mindestguthaben(140.0)
+ print(f"Kunden mit mind. 140€ Guthaben: {reiche_kunden}")
+
+ # 5. Kunde löschen
+ db.kunde_loeschen(id3)
+ print(f"Nach Löschen von Charlie: {db.alle_kunden()}")
+
+ db.schliessen()
+ print("Fertig!")
diff --git a/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/index.html b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/index.html
new file mode 100644
index 0000000..c5aff6f
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/index.html
@@ -0,0 +1,502 @@
+
+
+
+
+
+ Kapitel 21: Relationale Datenbanken & SQLite
+
+
+
+
+
+
+
+
+
+
+
+
Kapitel 21 • Data Engineering • SQL & Persistenz
+
Relationale Datenbanken & SQLite 🗄️⚡
+
Lerne die Grundlagen relationaler Datenbanken mit Pythons integriertem sqlite3-Modul, meistere CRUD-Operationen und schütze deine Daten vor SQL-Injections.
1. Was ist eine relationale Datenbank? (Die Aktenschrank-Analogie)
+
+
+ In vorherigen Kapiteln haben wir Daten in einfachen JSON- oder CSV-Dateien gespeichert. Bei größeren Anwendungen stoßen Textdateien jedoch schnell an ihre Grenzen:
+ Man kann nicht blitzschnell nach einzelnen Datensätzen suchen, ohne die gesamte Datei einzulesen, und gleichzeitige Schreibzugriffe führen zu Datenverlust.
+ Hier kommen relationale Datenbanksysteme (RDBMS) ins Spiel!
+
+
+
+
🗄️ Die Aktenschrank-Analogie:
+ Stell dir eine relationale Datenbank wie einen soliden Büroschrank vor:
+
+
Die Datenbankdatei (.db): Der gesamte Aktenschrank im Büro.
+
Die Tabelle (Table): Eine einzelne ausziehbare Schublade (z.B. Schublade „Kunden“).
+
Die Spalte (Column / Field): Ein vorgedrucktes Formularfeld auf den Karteikarten (z.B. „Name“, „E-Mail“, „Guthaben“).
+
Die Zeile (Row / Record): Eine einzelne, ausgefüllte Karteikarte für genau einen Kunden.
+
Der Primärschlüssel (PRIMARY KEY): Die eindeutige Kundennummer oben rechts auf jeder Karteikarte.
+
+
+
+
Der Lebenszyklus einer SQLite-Verbindung in Python
+
+ Python bringt mit sqlite3 eine vollständige, vollwertige SQL-Datenbank direkt in der Standardbibliothek mit – ohne Server-Installation!
+
+
+
+
+ Grundgerüst mit sqlite3
+
+
importsqlite3
+
+# 1. Verbindung herstellen (erstellt die Datei automatisch, falls nicht vorhanden)
+conn = sqlite3.connect("kunden.db")
+
+# 2. row_factory aktivieren: Erlaubt Dict-artigen Spaltenzugriff wie row["name"]
+conn.row_factory = sqlite3.Row
+
+# 3. Cursor erzeugen (der 'Stift', mit dem wir Befehle an die DB schicken)
+cursor = conn.cursor()
+
+# 4. SQL-Befehl ausführen
+cursor.execute("""
+ CREATE TABLE IF NOT EXISTS kunden (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ name TEXT NOT NULL,
+ email TEXT NOT NULL UNIQUE,
+ guthaben REAL DEFAULT 0.0
+ )
+""")
+
+# 5. Transaktion dauerhaft speichern (Commit)
+conn.commit()
+
+# 6. Verbindung sauber schließen
+conn.close()
+
+
+
+
+
+
2. Die 4 Säulen von SQL: CRUD im Überblick
+
In relationalen Datenbanken dreht sich alles um die vier elementaren CRUD-Operationen:
cursor.execute("SELECT * FROM kunden WHERE id = ?", (kunden_id,)) row = cursor.fetchone()
+
+
+
Update (Aktualisieren)
+
UPDATE ... SET
+
cursor.execute("UPDATE kunden SET guthaben = guthaben + ? WHERE id = ?", (betrag, kunden_id))
+
+
+
Delete (Löschen)
+
DELETE FROM ...
+
cursor.execute("DELETE FROM kunden WHERE id = ?", (kunden_id,))
+
+
+
+
+
+
💡 Cursor-Methoden im Detail:
+
+
cursor.fetchone(): Liefert die nächste einzelne Ergebniszeile (oder None).
+
cursor.fetchall(): Liefert eine Liste aller passenden Ergebniszeilen (oder []).
+
cursor.lastrowid: Die automatisch generierte ID des zuletzt eingefügten Datensatzes.
+
cursor.rowcount: Die Anzahl der Zeilen, die durch das letzte UPDATE oder DELETE verändert wurden.
+
+
+
+
+
+
+
3. 🚨 Sicherheit: Schutz vor SQL-Injections mit ?-Parametern
+
+
+ Eine der häufigsten und gefährlichsten Sicherheitslücken in der Software-Entwicklung ist die SQL-Injection.
+ Sie entsteht, wenn Benutzereingaben unbedacht per String-Formatierung (z.B. f-Strings) direkt in den SQL-Befehl eingebettet werden!
+
+
+
+
+
❌ GEFÄHRLICH: String-Formatierung
+
# Angreifer gibt als Name ein:
+# Robert'); DROP TABLE kunden;--
+
+query = f"INSERT INTO kunden (name) VALUES ('{user_name}')"
+cursor.execute(query)
+# BUMM! Die Tabelle kunden wird gelöscht!
+
+
+
+
✅ SICHER: Parametrisierte Abfrage
+
# Das sqlite3-Modul maskiert alle Zeichen:
+
+query = "INSERT INTO kunden (name) VALUES (?)"
+cursor.execute(query, (user_name,))
+
+# Der Name wird harmlos als Text gespeichert!
+
+
+
+
+
+
+
4. In-Memory Datenbanken mit :memory:
+
+ Für automatisierte Unit Tests oder temporäre Berechnungen bietet SQLite ein geniales Feature: die In-Memory-Datenbank.
+
+
+
conn = sqlite3.connect(":memory:")
+
+
+ Die Datenbank existiert ausschließlich im flüchtigen RAM. Sobald conn.close() aufgerufen wird oder der Test endet, ist alles rückstandslos bereinigt –
+ ideal für blitzschnelle, voneinander isolierte Tests ohne lästige temporäre Testdateien auf der Festplatte!
+
Öffne die Datei aufgabe.py und implementiere die Klasse KundenDatenbank:
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
💡 Lösungshinweise (Aufklappen bei Bedarf)
+
+
+ 🔍 Tipp zur Tabellenerstellung
+
+
+
self.cursor.execute("""
+ CREATE TABLE IF NOT EXISTS kunden (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ name TEXT NOT NULL,
+ email TEXT NOT NULL UNIQUE,
+ guthaben REAL DEFAULT 0.0
+ )
+""")
+self.conn.commit()
Führe im Terminal im Ordner 21_datenbanken_und_sqlite folgenden Befehl aus:
+
+
+ Terminal
+
+
python3 test_aufgabe.py
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/musterloesung.py b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/musterloesung.py
new file mode 100644
index 0000000..eb1e93b
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/musterloesung.py
@@ -0,0 +1,140 @@
+"""
+Kapitel 21: Relationale Datenbanken & SQLite (sqlite3) 🗄️⚡
+=========================================================
+Lehrpfad 4: Professional Data Engineering
+
+Musterlösung für die KundenDatenbank mit SQLite3.
+"""
+
+import sqlite3
+from typing import List, Optional, Dict, Any
+
+
+class KundenDatenbank:
+ """
+ Vollständig implementierte KundenDatenbank mit sqlite3.
+ """
+
+ def __init__(self, db_pfad: str = ":memory:"):
+ """
+ Initialisiert die Datenbankverbindung und erstellt die Tabelle 'kunden'.
+ """
+ self.db_pfad = db_pfad
+ self.conn = sqlite3.connect(db_pfad)
+ self.conn.row_factory = sqlite3.Row
+ self.cursor = self.conn.cursor()
+
+ # Tabelle anlegen
+ self.cursor.execute(
+ """
+ CREATE TABLE IF NOT EXISTS kunden (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ name TEXT NOT NULL,
+ email TEXT NOT NULL UNIQUE,
+ guthaben REAL DEFAULT 0.0
+ )
+ """
+ )
+ self.conn.commit()
+
+ def kunde_hinzufuegen(self, name: str, email: str, startguthaben: float = 0.0) -> int:
+ """
+ Fügt einen neuen Kunden per parametrisiertem INSERT ein.
+ """
+ self.cursor.execute(
+ "INSERT INTO kunden (name, email, guthaben) VALUES (?, ?, ?)",
+ (name, email, float(startguthaben))
+ )
+ self.conn.commit()
+ return self.cursor.lastrowid
+
+ def kunde_suchen_nach_id(self, kunden_id: int) -> Optional[Dict[str, Any]]:
+ """
+ Sucht einen Kunden anhand der ID und liefert ein Dictionary zurück.
+ """
+ self.cursor.execute(
+ "SELECT id, name, email, guthaben FROM kunden WHERE id = ?",
+ (kunden_id,)
+ )
+ row = self.cursor.fetchone()
+ if row is None:
+ return None
+ return {
+ "id": row["id"],
+ "name": row["name"],
+ "email": row["email"],
+ "guthaben": float(row["guthaben"]),
+ }
+
+ def guthaben_aufladen(self, kunden_id: int, betrag: float) -> bool:
+ """
+ Erhöht das Guthaben eines Kunden um einen positiven Betrag.
+ """
+ if betrag <= 0:
+ return False
+
+ self.cursor.execute(
+ "UPDATE kunden SET guthaben = guthaben + ? WHERE id = ?",
+ (float(betrag), kunden_id)
+ )
+ self.conn.commit()
+ return self.cursor.rowcount > 0
+
+ def kunden_mit_mindestguthaben(self, min_guthaben: float) -> List[Dict[str, Any]]:
+ """
+ Liefert alle Kunden mit mindestens min_guthaben, sortiert absteigend.
+ """
+ self.cursor.execute(
+ "SELECT id, name, email, guthaben FROM kunden WHERE guthaben >= ? ORDER BY guthaben DESC",
+ (float(min_guthaben),)
+ )
+ rows = self.cursor.fetchall()
+ return [
+ {
+ "id": r["id"],
+ "name": r["name"],
+ "email": r["email"],
+ "guthaben": float(r["guthaben"]),
+ }
+ for r in rows
+ ]
+
+ def kunde_loeschen(self, kunden_id: int) -> bool:
+ """
+ Löscht einen Kunden anhand der ID.
+ """
+ self.cursor.execute(
+ "DELETE FROM kunden WHERE id = ?",
+ (kunden_id,)
+ )
+ self.conn.commit()
+ return self.cursor.rowcount > 0
+
+ def alle_kunden(self) -> List[Dict[str, Any]]:
+ """
+ Gibt eine Liste aller Kunden zurück.
+ """
+ self.cursor.execute("SELECT id, name, email, guthaben FROM kunden ORDER BY id ASC")
+ rows = self.cursor.fetchall()
+ return [
+ {
+ "id": r["id"],
+ "name": r["name"],
+ "email": r["email"],
+ "guthaben": float(r["guthaben"]),
+ }
+ for r in rows
+ ]
+
+ def schliessen(self) -> None:
+ """Schließt die Verbindung."""
+ if hasattr(self, "conn") and self.conn:
+ self.conn.close()
+
+
+if __name__ == "__main__":
+ db = KundenDatenbank(":memory:")
+ i1 = db.kunde_hinzufuegen("Max Mustermann", "max@test.de", 100.0)
+ print("Kunde hinzugefügt mit ID:", i1)
+ print("Kunde gesucht:", db.kunde_suchen_nach_id(i1))
+ db.schliessen()
diff --git a/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/test_aufgabe.py b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/test_aufgabe.py
new file mode 100644
index 0000000..84cf9ec
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/21_datenbanken_und_sqlite/test_aufgabe.py
@@ -0,0 +1,156 @@
+import unittest
+import sys
+import sqlite3
+from pathlib import Path
+
+# Sicherstellen, dass das aktuelle Verzeichnis im Python-Pfad liegt
+sys.path.insert(0, str(Path(__file__).parent))
+
+try:
+ import aufgabe
+except ImportError as e:
+ raise ImportError(f"Konnte 'aufgabe.py' nicht importieren: {e}")
+
+
+class TestKapitel21(unittest.TestCase):
+ """
+ Umfassende Testsuite für Modul 21 (SQLite Datenbanken & Persistenz).
+ Nutzt isolierte In-Memory-Datenbanken (:memory:) für maximale Geschwindigkeit.
+ """
+
+ def setUp(self):
+ self.KundenDatenbank = getattr(aufgabe, "KundenDatenbank", None)
+ self.assertIsNotNone(self.KundenDatenbank, "Klasse 'KundenDatenbank' nicht in aufgabe.py gefunden!")
+ # Jede Testmethode erhält eine frische, isolierte In-Memory-Datenbank
+ self.db = self.KundenDatenbank(":memory:")
+
+ def tearDown(self):
+ if hasattr(self, "db") and self.db:
+ self.db.schliessen()
+
+ def test_01_tabelle_und_initialisierung(self):
+ """Prüft, ob die Tabelle 'kunden' mit den korrekten Spalten angelegt wurde."""
+ cursor = self.db.conn.cursor()
+ cursor.execute("PRAGMA table_info(kunden)")
+ spalten = {row[1]: row[2].upper() for row in cursor.fetchall()}
+
+ self.assertIn("id", spalten, "Spalte 'id' fehlt in Tabelle 'kunden'!")
+ self.assertIn("name", spalten, "Spalte 'name' fehlt in Tabelle 'kunden'!")
+ self.assertIn("email", spalten, "Spalte 'email' fehlt in Tabelle 'kunden'!")
+ self.assertIn("guthaben", spalten, "Spalte 'guthaben' fehlt in Tabelle 'kunden'!")
+
+ def test_02_kunde_hinzufuegen(self):
+ """Prüft das Einfügen von Kunden und die Rückgabe der neuen ID."""
+ id1 = self.db.kunde_hinzufuegen("Alice Schmidt", "alice@example.com", 150.0)
+ self.assertIsInstance(id1, int, "kunde_hinzufuegen muss die int-ID zurückgeben!")
+ self.assertEqual(id1, 1, "Der erste Kunde sollte ID 1 erhalten!")
+
+ id2 = self.db.kunde_hinzufuegen("Bob Müller", "bob@example.com", 75.50)
+ self.assertEqual(id2, 2, "Der zweite Kunde sollte ID 2 erhalten!")
+
+ # Prüfen, ob Datensätze direkt in der DB liegen
+ c = self.db.conn.cursor()
+ c.execute("SELECT COUNT(*) FROM kunden")
+ anzahl = c.fetchone()[0]
+ self.assertEqual(anzahl, 2, "Es sollten genau 2 Kunden in der Datenbank vorhanden sein!")
+
+ def test_03_kunde_suchen_nach_id(self):
+ """Prüft die Suche nach existierenden und nicht-existierenden Kunden."""
+ id_neu = self.db.kunde_hinzufuegen("Clara Oswald", "clara@tardis.org", 250.0)
+
+ # Existierender Kunde
+ kunde = self.db.kunde_suchen_nach_id(id_neu)
+ self.assertIsInstance(kunde, dict, "kunde_suchen_nach_id muss ein Dict zurückgeben!")
+ self.assertEqual(kunde["id"], id_neu)
+ self.assertEqual(kunde["name"], "Clara Oswald")
+ self.assertEqual(kunde["email"], "clara@tardis.org")
+ self.assertAlmostEqual(kunde["guthaben"], 250.0, places=2)
+
+ # Nicht existierender Kunde
+ unbekannt = self.db.kunde_suchen_nach_id(9999)
+ self.assertIsNone(unbekannt, "kunde_suchen_nach_id muss None zurückgeben, wenn ID nicht existiert!")
+
+ def test_04_guthaben_aufladen(self):
+ """Prüft das Aufladen von Guthaben und Validierung von Beträgen."""
+ kid = self.db.kunde_hinzufuegen("David Miller", "david@web.de", 100.0)
+
+ # Gültiges Aufladen
+ erfolg = self.db.guthaben_aufladen(kid, 50.0)
+ self.assertTrue(erfolg, "guthaben_aufladen sollte True zurückgeben bei erfolgreicher Aktualisierung!")
+
+ kunde = self.db.kunde_suchen_nach_id(kid)
+ self.assertAlmostEqual(kunde["guthaben"], 150.0, places=2)
+
+ # Ungültiger Betrag (<= 0)
+ erfolg_negativ = self.db.guthaben_aufladen(kid, -20.0)
+ self.assertFalse(erfolg_negativ, "guthaben_aufladen darf bei Beträgen <= 0 keine Buchung durchführen (False)!")
+
+ erfolg_null = self.db.guthaben_aufladen(kid, 0.0)
+ self.assertFalse(erfolg_null, "guthaben_aufladen darf bei Betrag 0.0 nicht erfolgreich sein!")
+
+ # Nicht-existierende Kunden-ID
+ erfolg_fremd = self.db.guthaben_aufladen(8888, 50.0)
+ self.assertFalse(erfolg_fremd, "guthaben_aufladen sollte False zurückgeben, wenn Kunden-ID nicht existiert!")
+
+ def test_05_kunden_mit_mindestguthaben(self):
+ """Prüft die Abfrage und Sortierung von Kunden nach Mindestguthaben."""
+ self.db.kunde_hinzufuegen("Sparfuchs", "sparfuchs@bank.de", 10.0)
+ self.db.kunde_hinzufuegen("Mittelstand", "mittel@bank.de", 500.0)
+ self.db.kunde_hinzufuegen("Vip Kunde", "vip@bank.de", 1200.0)
+ self.db.kunde_hinzufuegen("Grossanleger", "investor@bank.de", 2500.0)
+
+ # Filter ab 500.0
+ ergebnis = self.db.kunden_mit_mindestguthaben(500.0)
+ self.assertIsInstance(ergebnis, list, "kunden_mit_mindestguthaben muss eine Liste zurückgeben!")
+ self.assertEqual(len(ergebnis), 3, "Es sollten genau 3 Kunden >= 500€ Guthaben haben!")
+
+ # Prüfung der absteigenden Sortierung (Grossanleger > Vip Kunde > Mittelstand)
+ self.assertEqual(ergebnis[0]["name"], "Grossanleger")
+ self.assertEqual(ergebnis[1]["name"], "Vip Kunde")
+ self.assertEqual(ergebnis[2]["name"], "Mittelstand")
+
+ # Zu hohe Schwelle -> leere Liste
+ keine = self.db.kunden_mit_mindestguthaben(10000.0)
+ self.assertEqual(keine, [], "Sollte leere Liste zurückgeben, wenn kein Kunde die Schwelle erreicht!")
+
+ def test_06_kunde_loeschen(self):
+ """Prüft das Löschen von Kunden aus der Datenbank."""
+ kid = self.db.kunde_hinzufuegen("Ex-Kunde", "ex@alt.de", 0.0)
+
+ loesch_erfolg = self.db.kunde_loeschen(kid)
+ self.assertTrue(loesch_erfolg, "kunde_loeschen sollte True zurückgeben, wenn Kunde existierte!")
+
+ # Erneute Suche muss None liefern
+ kunde = self.db.kunde_suchen_nach_id(kid)
+ self.assertIsNone(kunde, "Gelöschter Kunde darf über kunde_suchen_nach_id nicht mehr gefunden werden!")
+
+ # Zweites Löschen derselben ID muss False liefern
+ loesch_nochmal = self.db.kunde_loeschen(kid)
+ self.assertFalse(loesch_nochmal, "kunde_loeschen sollte False liefern, wenn ID nicht existiert!")
+
+ def test_07_sql_injection_schutz(self):
+ """Prüft, ob parametrisierte Queries vor böswilligen SQL-Injections schützen."""
+ boeser_name = "Robert'); DROP TABLE kunden;--"
+ kid = self.db.kunde_hinzufuegen(boeser_name, "bobby@tables.com", 100.0)
+
+ kunde = self.db.kunde_suchen_nach_id(kid)
+ self.assertIsNotNone(kunde, "Kunde mit SQL-Injection-String im Namen sollte als regulärer Text gespeichert werden!")
+ self.assertEqual(kunde["name"], boeser_name)
+
+ # Tabelle 'kunden' muss intakt sein!
+ c = self.db.conn.cursor()
+ c.execute("SELECT COUNT(*) FROM kunden")
+ self.assertEqual(c.fetchone()[0], 1, "Tabelle 'kunden' wurde trotz SQL-Injection-Versuch nicht zerstört!")
+
+
+if __name__ == "__main__":
+ res = unittest.main(verbosity=2, exit=False)
+ try:
+ root_dir = Path(__file__).parent.parent.parent.resolve()
+ sys.path.insert(0, str(root_dir))
+ from gamification import report_single_chapter_result
+ rel_pfad = f"{Path(__file__).parent.parent.name}/{Path(__file__).parent.name}"
+ report_single_chapter_result(rel_pfad, res.result.wasSuccessful(), res.result.testsRun)
+ except Exception:
+ pass
+ sys.exit(0 if res.result.wasSuccessful() else 1)
diff --git a/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/README.md b/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/README.md
new file mode 100644
index 0000000..93e04fd
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/README.md
@@ -0,0 +1,48 @@
+# Kapitel 22: Web-APIs, REST & JSON-Feeds 🌐📡
+
+In diesem Kapitel lernst du, wie moderne Web-Anwendungen über REST-APIs kommunizieren, wie du strukturierte JSON-Feeds in Python parst, verschachtelte Daten extrahierst und Fehler robust abfängst.
+
+---
+
+## 🎯 Was du lernst
+
+1. **Was ist eine REST-API?**
+ - Client-Server-Architektur im Web: Der Client stellt eine Anfrage (**Request**) per HTTP GET/POST an den Endpunkt, der Server antwortet (**Response**).
+2. **Die Kellner-Analogie:**
+ - **Gast (Client/Python-Skript):** Möchte Daten (z.B. Wetterbericht oder Aktienkurse) haben.
+ - **Kellner (API):** Nimmt die Bestellung entgegen, bringt sie in die Küche und serviert die Antwort.
+ - **Küche/Koch (Server/Datenbank):** Verarbeitet die Anfrage und bereitet die Rohdaten zu.
+ - **Speisekarte (API-Dokumentation):** Beschreibt, welche Endpunkte und Parameter erlaubt sind.
+3. **JSON als Universalsprache des Internets:**
+ - `json.loads(json_string)`: Wandelt einen empfangenen JSON-Text in ein Python-Dictionary bzw. eine Python-Liste um.
+ - `json.dumps(obj, indent=2)`: Wandelt ein Python-Objekt in einen JSON-formatierten String um.
+4. **HTTP Status-Codes:**
+ - `200 OK`: Anfrage erfolgreich beantwortet.
+ - `400 Bad Request`: Fehlerhafte Anfrage / ungültige Parameter.
+ - `404 Not Found`: Endpunkt oder Ressource nicht gefunden.
+ - `500 Internal Server Error`: Unerwarteter Serverfehler.
+5. **Robuste Fehlerbehandlung & Defensives Parsen:**
+ - `json.JSONDecodeError`: Wenn der Server HTML statt JSON sendet oder der Payload unvollständig ist.
+ - `KeyError` / `IndexError`: Wenn erwartete Felder im JSON fehlen.
+ - Sicheres Auslesen mit `.get()` und Standardwerten.
+6. **100% Offline- & Pyodide-kompatibel:**
+ - Alle Übungen und Tests nutzen realistische Mock-JSON-Strings, sodass keine Internetverbindung oder externe API-Keys benötigt werden.
+
+---
+
+## 📁 Die Dateien in diesem Ordner
+
+- **`index.html`**: Die interaktive Lernseite mit Kellner-Analogie, API-Ablaufdiagramm, JSON Cheat-Sheet und Subgoals.
+- **`aufgabe.py`**: Dein Arbeitsblatt mit Parser-Funktionen und der Klasse `WetterApiParser`.
+- **`test_aufgabe.py`**: Automatische Unittest-Suite (`python3 test_aufgabe.py`).
+- **`musterloesung.py`**: Vollständig ausprogrammierte Beispiellösung.
+
+---
+
+## 🚀 Schnellstart
+
+```bash
+# 1. Bearbeite aufgabe.py
+# 2. Teste deine Lösung:
+python3 test_aufgabe.py
+```
diff --git a/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/aufgabe.py b/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/aufgabe.py
new file mode 100644
index 0000000..b468539
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/aufgabe.py
@@ -0,0 +1,222 @@
+"""
+Kapitel 22: Web-APIs, REST & JSON-Feeds 🌐📡
+============================================
+Lehrpfad 4: Professional Data Engineering
+
+Thema:
+REST-APIs, JSON als Datenformat, strukturierte Extraktion verschachtelter Payloads
+und defensive Fehlerbehandlung (KeyError, JSONDecodeError).
+
+Lernziele:
+1. JSON-Strings parsen mit `json.loads(text)` und erzeugen mit `json.dumps(obj)`.
+2. Extrahieren von Werten aus mehrfach verschachtelten Dictionaries und Listen.
+3. Filtern und Transformieren von Web-Feeds (z.B. Aktienkurse, News-Ticker).
+4. Validieren von HTTP-Statuscodes (200 OK vs. 404/500) und Fehler-Payloads.
+5. Kapselung der API-Logik in einer wiederverwendbaren Klasse `WetterApiParser`.
+"""
+
+import json
+from typing import List, Dict, Any, Tuple, Optional, Union
+
+
+# ==============================================================================
+# 🎯 TEILZIEL 1 (TODO 1): Wetter-Antwort parsen
+# ==============================================================================
+def parse_wetter_antwort(json_text: str) -> Dict[str, Any]:
+ """
+ Parst den JSON-String einer Wetter-API (z.B. OpenWeatherMap-Format)
+ und extrahiert die wichtigsten Kenngrößen in ein kompaktes Dictionary.
+
+ Erwartete Eingabe-Struktur (Beispiel):
+ {
+ "name": "Wien",
+ "main": {
+ "temp": 21.5,
+ "humidity": 65
+ },
+ "weather": [
+ {
+ "main": "Clouds",
+ "description": "Leicht bewölkt"
+ }
+ ]
+ }
+
+ Rückgabe-Format:
+ {
+ "stadt": str, # aus data["name"]
+ "temperatur": float, # aus data["main"]["temp"]
+ "luftfeuchtigkeit": int, # aus data["main"]["humidity"]
+ "wetterlage": str # aus data["weather"][0]["description"] (oder data["weather"][0]["main"])
+ }
+
+ Fehlerbehandlung:
+ - Wenn json_text ungültiges JSON ist (json.JSONDecodeError)
+ oder erwartete Schlüssel fehlen (KeyError, IndexError, TypeError):
+ Wirf einen ValueError("Ungültige Wetterdaten: " + str(e))
+ """
+ # TODO: Implementieren
+ pass
+
+
+# ==============================================================================
+# 🎯 TEILZIEL 2 (TODO 2): Aktienkurse aus JSON-Feed filtern
+# ==============================================================================
+def filtriere_aktien_kurse(json_feed_text: str, min_kurs: float) -> List[Dict[str, Any]]:
+ """
+ Parst einen Finanzmarkt-Feed und filtert alle Aktien heraus, deren Kurs
+ größer oder gleich `min_kurs` ist.
+
+ Akzeptiert Formate:
+ 1. {"aktien": [{"symbol": "AAPL", "kurs": 185.5, "waehrung": "USD"}, ...]}
+ 2. oder direkt eine Liste: [{"symbol": "AAPL", "kurs": 185.5}, ...]
+
+ Rückgabewert:
+ - Eine Liste von Dictionaries der gefilterten Aktien (z.B. [ {"symbol": "AAPL", "kurs": 185.5, ...}, ... ])
+ - Bei ungültigem JSON oder leerem Feed: leere Liste [] zurückgeben.
+ """
+ # TODO: Implementieren
+ pass
+
+
+# ==============================================================================
+# 🎯 TEILZIEL 3 (TODO 3): Schlagzeilen-Ticker formatieren
+# ==============================================================================
+def formatiere_nachrichten_ticker(news_json_text: str) -> str:
+ """
+ Parst einen News-JSON-Feed und formatiert die Schlagzeilen zu einem durchgehenden
+ Laufband-String für einen Nachrichtenticker.
+
+ Erwartete Eingabe-Struktur:
+ {
+ "nachrichten": [
+ {"titel": "Python 3.13 veröffentlicht", "ressort": "Tech"},
+ {"titel": "DAX schließt im Plus", "ressort": "Wirtschaft"}
+ ]
+ }
+
+ Formatierungsregel:
+ - Jede Nachricht wird formatiert als: "[] +++ "
+ - Beispiel-Ergebnis:
+ "[Tech] Python 3.13 veröffentlicht +++ [Wirtschaft] DAX schließt im Plus +++"
+ - Falls keine Nachrichten vorhanden sind oder die Liste leer ist:
+ "+++ Keine aktuellen Eilmeldungen +++" zurückgeben.
+ - Bei ungültigem JSON: "+++ Fehler beim Laden des Nachrichten-Feeds +++" zurückgeben.
+ """
+ # TODO: Implementieren
+ pass
+
+
+# ==============================================================================
+# 🎯 TEILZIEL 4 (TODO 4): HTTP-Antwort & Statuscode validieren
+# ==============================================================================
+def validiere_api_antwort(status_code: int, payload_json_text: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
+ """
+ Überprüft den HTTP-Statuscode und den JSON-Payload einer Web-API-Antwort.
+
+ Regeln:
+ 1. Wenn status_code == 200:
+ - Versuche payload_json_text mit json.loads() zu parsen.
+ - Bei Erfolg: Gib (True, geparstes_dict) zurück.
+ - Bei json.JSONDecodeError: Gib (False, "Ungültiges JSON-Format") zurück.
+
+ 2. Wenn status_code != 200 (z.B. 400, 404, 500):
+ - Versuche aus dem JSON-Payload eine Fehlermeldung zu extrahieren (Schlüssel 'error' oder 'message').
+ - Falls vorhanden: Gib (False, f"HTTP Fehler {status_code}: {fehlermeldung}") zurück.
+ - Falls nicht parsebar oder kein Schlüssel: Gib (False, f"HTTP Fehler {status_code}") zurück.
+ """
+ # TODO: Implementieren
+ pass
+
+
+# ==============================================================================
+# 🎯 TEILZIEL 5 (TODO 5): Klasse WetterApiParser
+# ==============================================================================
+class WetterApiParser:
+ """
+ Objektorientierter Wrapper für Wetterdaten aus einer REST-API.
+ """
+
+ def __init__(self, raw_json: str = ""):
+ """
+ Initialisiert den Parser.
+ Falls raw_json übergeben wurde, wird sofort lade_daten() ausgeführt.
+ """
+ self.raw_json: str = raw_json
+ self.daten: Optional[Dict[str, Any]] = None
+ if raw_json:
+ self.lade_daten(raw_json)
+
+ def lade_daten(self, raw_json: str) -> bool:
+ """
+ Versucht den JSON-String mittels parse_wetter_antwort() einzulesen.
+ - Bei Erfolg: Speichert das extrahierte Dict in self.daten und gibt True zurück.
+ - Bei Fehler (ValueError): Setzt self.daten = None und gibt False zurück.
+ """
+ # TODO: Implementieren
+ pass
+
+ def get_temperatur(self) -> Optional[float]:
+ """Gibt die Temperatur zurück oder None, wenn keine Daten geladen sind."""
+ # TODO: Implementieren
+ pass
+
+ def get_stadt(self) -> Optional[str]:
+ """Gibt den Stadtnamen zurück oder None, wenn keine Daten geladen sind."""
+ # TODO: Implementieren
+ pass
+
+ def ist_frostig(self) -> bool:
+ """Gibt True zurück, wenn die Temperatur unter 0.0 °C liegt, sonst False."""
+ # TODO: Implementieren
+ pass
+
+ def ist_gueltig(self) -> bool:
+ """Gibt True zurück, wenn gültige Wetterdaten geladen sind (self.daten is not None)."""
+ # TODO: Implementieren
+ pass
+
+
+# ==============================================================================
+# Interaktiver Test im Terminal:
+# (python3 aufgabe.py)
+# ==============================================================================
+if __name__ == "__main__":
+ beispiel_wetter = """
+ {
+ "name": "München",
+ "main": {"temp": 18.2, "humidity": 72},
+ "weather": [{"main": "Rain", "description": "Leichter Regen"}]
+ }
+ """
+ print("🌤️ Teste parse_wetter_antwort():")
+ wetter = parse_wetter_antwort(beispiel_wetter)
+ print(wetter)
+
+ beispiel_aktien = """
+ {
+ "aktien": [
+ {"symbol": "AAPL", "kurs": 189.5, "waehrung": "USD"},
+ {"symbol": "GOOGL", "kurs": 142.0, "waehrung": "USD"},
+ {"symbol": "PENNY", "kurs": 0.45, "waehrung": "USD"}
+ ]
+ }
+ """
+ print("\n📈 Teste filtriere_aktien_kurse() (min_kurs = 100):")
+ gefiltert = filtriere_aktien_kurse(beispiel_aktien, 100.0)
+ print(gefiltert)
+
+ beispiel_news = """
+ {
+ "nachrichten": [
+ {"titel": "Python 3.13 bringt Performance-Schub", "ressort": "Technologie"},
+ {"titel": "Mars-Rover findet neue Gesteinsproben", "ressort": "Wissenschaft"}
+ ]
+ }
+ """
+ print("\n📰 Teste formatiere_nachrichten_ticker():")
+ print(formatiere_nachrichten_ticker(beispiel_news))
+
+ print("\n🛡️ Teste validiere_api_antwort():")
+ print("Status 200:", validiere_api_antwort(200, '{"status": "ok", "items": [1, 2, 3]}'))
+ print("Status 404:", validiere_api_antwort(404, '{"error": "Ressource nicht gefunden"}'))
diff --git a/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/index.html b/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/index.html
new file mode 100644
index 0000000..b7a91ed
--- /dev/null
+++ b/lehrpfad_4_professional_data_engineering/22_web_apis_und_json_feeds/index.html
@@ -0,0 +1,443 @@
+
+
+
+
+
+ Kapitel 22: Web-APIs, REST & JSON-Feeds
+
+
+
+
+
+
+
+
+
+
+
+
Kapitel 22 • Data Engineering • Web & REST APIs
+
Web-APIs, REST & JSON-Feeds 🌐📡
+
Lerne, wie Web-APIs funktionieren, wie du strukturierte JSON-Feeds in Python parst, Daten filterst und Netzwerk- sowie Parsing-Fehler professionell abfängst.
+ Eine API (Application Programming Interface) ist eine Schnittstelle, über die zwei Computer oder Programme übers Internet miteinander kommunizieren.
+ Nahezu jede moderne Web-App (wie Wetterberichte, Börsenkurse, Spotify oder Google Maps) stellt ihre Daten über REST-APIs im JSON-Format bereit.
+
+
+
+
🍽️ Die Restaurant-Analogie:
+ Stell dir den Ablauf in einem Restaurant vor:
+
+
Der Gast (Client / Dein Python-Skript): Du sitzt am Tisch und möchtest ein bestimmtes Gericht (Daten) bestellen.
+
Die Speisekarte (API-Dokumentation / Endpunkte): Zeigt dir, welche Gerichte (URLs & Parameter wie /weather?city=Wien) du bestellen kannst.
+
Der Kellner (Die REST-API): Nimmt deine Bestellung (**HTTP Request**) entgegen, eilt in die Küche und bringt dir das fertige Essen (**HTTP Response**) zurück.
+
Die Küche & Vorratskammer (Web-Server & Datenbank): Verarbeitet die Rohzutaten und bereitet das Essen zu.
+
+
+
+
Der typische Ablauf eines API-Aufrufs
+
+
[Client: Python-App]
+
│ 🚀 1. HTTP GET Request (z.B. https://api.wetter.com/v1/forecast?city=Berlin)