import sqlite3
from typing import Optional, List, Any
DB_PATH = "rezepte.db"
def verbindung_herstellen(path: str = DB_PATH) -> sqlite3.Connection:
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
return conn
def erstelle_tabellen(conn: sqlite3.Connection) -> None:
sql = """
CREATE TABLE IF NOT EXISTS recipes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
ingredients TEXT,
steps TEXT,
servings INTEGER,
prep_time_minutes INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""
conn.execute(sql)
conn.commit()
def beispieldaten_einfuegen(conn: sqlite3.Connection) -> None:
"""bsp daten einfügen (idempotent nicht geprüft lol)."""
samples = [
("Schnelle Tomatensauce",
"Tomaten\nZwiebel\nKnoblauch\nOlivenöl\nSalz\nPfeffer\nBasilikum",
"Zwiebel+Knoblauch anbraten\nTomaten hinzufügen 15 min köcheln\nabschmecken",
4, 20),
("Rührei mit Kräutern",
"Eier\nMilch\nButter\nKräuter\nSalz\nPfeffer",
"Eier verquirlen\nin Butter stocken lassen\nKräuter unterheben",
1, 10),
]
sql = "INSERT INTO recipes (name, ingredients, steps, servings, prep_time_minutes) VALUES (?, ?, ?, ?, ?)"
conn.executemany(sql, samples)
conn.commit()
def hole_alle_rezepte(conn: sqlite3.Connection) -> List[sqlite3.Row]:
"""alle Rezepte abrufen."""
cur = conn.execute("SELECT * FROM recipes ORDER BY created_at DESC")
return cur.fetchall()
def hole_rezept_nach_id(conn: sqlite3.Connection, recipe_id: int) -> Optional[sqlite3.Row]:
"""ein Rezept nach ID holen."""
cur = conn.execute("SELECT * FROM recipes WHERE id = ?", (recipe_id,))
return cur.fetchone()
def suche_rezepte_nach_zutat(conn: sqlite3.Connection, ingredient: str) -> List[sqlite3.Row]:
"""rezepte finden, die eine Zutat (Teilstring) enthalten."""
pattern = f"%{ingredient}%"
cur = conn.execute("SELECT * FROM recipes WHERE ingredients LIKE ? ORDER BY name", (pattern,))
return cur.fetchall()
def aktualisiere_rezept(conn: sqlite3.Connection, recipe_id: int, **fields: Any) -> bool:
"""
rezeptfelder aktualisieren. Beispiel:
aktualisiere_rezept(conn, 1, name="Neu", servings=2)
gibt True zurück, wenn ein Datensatz geändert wurde.
"""
allowed = {"name", "ingredients", "steps", "servings", "prep_time_minutes"}
set_parts = []
values = []
for k, v in fields.items():
if k in allowed:
set_parts.append(f"{k} = ?")
values.append(v)
if not set_parts:
return False
values.append(recipe_id)
sql = f"UPDATE recipes SET {', '.join(set_parts)} WHERE id = ?"
cur = conn.execute(sql, values)
conn.commit()
return cur.rowcount > 0
def loesche_rezept(conn: sqlite3.Connection, recipe_id: int) -> bool:
"""ein Rezept löschen. gibt TRUE zurück, wenn etwas gelöscht wurde. ^^"""
cur = conn.execute("DELETE FROM recipes WHERE id = ?", (recipe_id,))
conn.commit()
return cur.rowcount > 0
def drucke_rezepte(rows: List[sqlite3.Row]) -> None:
"""gebe zeugsl im lesbarem format aus."""
for r in rows:
print(f"[{r['id']}] {r['name']} — Portionen: {r['servings']}, Zeit: {r['prep_time_minutes']}min")
print(" Zutaten:")
for line in (r["ingredients"] or "").splitlines():
print(f" - {line}")
print(" Schritte:")
for line in (r["steps"] or "").splitlines():
print(f" {line}")
print("-" * 40)
if __name__ == "__main__":
conn = verbindung_herstellen()
erstelle_tabellen(conn)
# bsp befüllen
beispieldaten_einfuegen(conn)
# auslesen und anzeigen
alle = hole_alle_rezepte(conn)
print("Alle Rezepte:")
drucke_rezepte(alle)
# bsp: Update eines Eintrags (falls vorhanden(?))
if alle:
first_id = alle[0]["id"]
aktualisiere_rezept(conn, first_id, servings=2, prep_time_minutes=15)
print("\nNach Update:")
drucke_rezepte([hole_rezept_nach_id(conn, first_id)])
# bsp: Löschen eines Eintrags (falls vorhanden(?))
if alle:
loesche_rezept(conn, alle[-1]["id"])
print("\nNach Löschen letzter Eintrag:")
drucke_rezepte(hole_alle_rezepte(conn))
conn.close()
"""ich liebe docstrings <3"""