SemaTrain Logo Ein Fachportal von SemaTrain

Datenbanken mit SQLAlchemy: ORM sauber einsetzen

Sie speichern und lesen Daten sauber aus einer Datenbank: SQLAlchemy (ORM), Sessions, Modelle, CRUD und Queries – mit einem Mini-Beispiel im SemaTrain-Schulungskontext.

Hinweis: Wir starten mit SQLite (Datei-DB) – der Code ist später 1:1 auf PostgreSQL übertragbar (Connection-String tauschen).

Python Schulung – Kursbezug

Dieses Kapitel ist Teil des Lernpfads zur Python Schulung. Termine & Buchung laufen über SemaTrain.de.

Ziel: Sie bauen ein kleines ORM-Setup (SQLite), führen CRUD aus und formulieren Queries – als Basis für Web-Apps und Datenpipelines.

Merksatz: Session = Arbeitseinheit. Erst commit() macht Änderungen dauerhaft.

Worum geht’s?

Lehr-/Lernziele

Nach diesem Kapitel können Sie …

SemaTrain-Beispiel: Wir speichern „Kurse“ und „Teilnehmer“ (Beispieldaten) in SQLite. Später kann das Schema leicht erweitert werden (Trainer, Termine, Standorte, Buchungen).

Setup: Engine + Session (SQLite zuerst)

Engine + Session (LZ1) (Python)
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite (Datei) – perfekt für Lernpfad/Prototyp
engine = create_engine("sqlite:///daten/sematrain.db", echo=False)

Session = sessionmaker(bind=engine)

with Session() as session:
    # hier würden wir arbeiten (Queries/CRUD)
    print("Session bereit:", session is not None)

# PostgreSQL (Beispiel):
# engine = create_engine("postgresql+psycopg2://user:pass@localhost:5432/sematrain", echo=False)

ORM-Modelle: Kurs & Teilnehmer

Wir nutzen SQLAlchemy 2.x-Style (Mapped/relationship). Fokus: klar, robust, gut erweiterbar.

ORM-Modelle + Relationship (LZ2) (Python)
from sqlalchemy import ForeignKey, String, Integer, Float
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Kurs(Base):
    __tablename__ = "kurs"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    kursname: Mapped[str] = mapped_column(String(120), nullable=False)
    format: Mapped[str] = mapped_column(String(30), nullable=False)       # "Online" / "Praesenz"
    standort: Mapped[str] = mapped_column(String(60), nullable=False)     # z.B. "Hamburg"
    dauer_tage: Mapped[int] = mapped_column(Integer, nullable=False)
    preis_euro: Mapped[float] = mapped_column(Float, nullable=False)

    teilnehmer: Mapped[list["Teilnehmer"]] = relationship(
        back_populates="kurs",
        cascade="all, delete-orphan"
    )

class Teilnehmer(Base):
    __tablename__ = "teilnehmer"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(120), nullable=False)
    firma: Mapped[str] = mapped_column(String(120), nullable=False)

    kurs_id: Mapped[int] = mapped_column(ForeignKey("kurs.id"), nullable=False)
    kurs: Mapped["Kurs"] = relationship(back_populates="teilnehmer")
Tabellen erzeugen (create_all)
Base.metadata.create_all(engine) (LZ2) (Python)
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Stelle sicher, dass der Ordner existiert (SQLite-Datei)
Path("daten").mkdir(exist_ok=True)

engine = create_engine("sqlite:///daten/sematrain.db", echo=False)

# Importiere Base + Modelle (hier nur symbolisch)
# from modelle import Base, Kurs, Teilnehmer

# Base.metadata.create_all(engine)
print("Tabellen erzeugt (wenn Base + Modelle importiert sind).")

Tipp: In echten Projekten nutzt man für Schema-Änderungen meist Alembic (Migrationen) statt ständig create_all.

CRUD: Anlegen, Lesen, Aktualisieren, Löschen

CRUD mit commit() (LZ3) (Python)
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

# angenommen: Base, Kurs, Teilnehmer sind definiert
engine = create_engine("sqlite:///daten/sematrain.db", echo=False)
Session = sessionmaker(bind=engine)

with Session() as session:
    # CREATE (LZ3)
    kurs = Kurs(
        kursname="Python Schulung",
        format="Online",
        standort="Berlin",
        dauer_tage=3,
        preis_euro=1390.00
    )
    kurs.teilnehmer.append(Teilnehmer(name="Max Beispiel", firma="Beispiel GmbH"))
    kurs.teilnehmer.append(Teilnehmer(name="Ada Muster", firma="Muster AG"))

    session.add(kurs)
    session.commit()   # wichtig!

with Session() as session:
    # READ (LZ3)
    stmt = select(Kurs).where(Kurs.standort == "Berlin").order_by(Kurs.preis_euro.desc())
    kurse = session.execute(stmt).scalars().all()
    for k in kurse:
        print(k.kursname, k.format, k.preis_euro, "TN:", len(k.teilnehmer))

with Session() as session:
    # UPDATE (LZ3)
    kurs = session.execute(select(Kurs).where(Kurs.standort == "Berlin")).scalars().first()
    if kurs:
        kurs.preis_euro = 1290.00
        session.commit()

with Session() as session:
    # DELETE (LZ3)
    kurs = session.execute(select(Kurs).where(Kurs.standort == "Berlin")).scalars().first()
    if kurs:
        session.delete(kurs)
        session.commit()
Robust: rollback() bei Fehlern
try/except (Rollback-Prinzip) (LZ3) (Python)
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///daten/sematrain.db", echo=False)
Session = sessionmaker(bind=engine)

try:
    with Session() as session:
        kurs = Kurs(
            kursname="Python Schulung",
            format="Online",
            standort="Hamburg",
            dauer_tage=3,
            preis_euro=1490.00
        )
        session.add(kurs)
        session.commit()
except Exception as fehler:
    # Falls du Session außerhalb des with nutzt: session.rollback()
    print("Fehler – Transaktion abgebrochen:", fehler)

Queries: Filtern, Sortieren, Join über Relationship

Filter/Sort/Aggregation/Join (LZ4) (Python)
from sqlalchemy import create_engine, select, func
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///daten/sematrain.db", echo=False)
Session = sessionmaker(bind=engine)

with Session() as session:
    # Alle Online-Kurse, sortiert nach Preis
    stmt = select(Kurs).where(Kurs.format == "Online").order_by(Kurs.preis_euro.asc())
    for k in session.execute(stmt).scalars():
        print(k.standort, k.preis_euro)

with Session() as session:
    # Aggregation: Durchschnittspreis pro Format
    stmt = (
        select(Kurs.format, func.avg(Kurs.preis_euro))
        .group_by(Kurs.format)
    )
    for format_, avg_preis in session.execute(stmt).all():
        print(format_, round(avg_preis, 2))

with Session() as session:
    # Join via Relationship: Teilnehmer + Kursname
    stmt = (
        select(Teilnehmer.name, Kurs.kursname, Kurs.standort)
        .join(Teilnehmer.kurs)
        .order_by(Kurs.standort.asc(), Teilnehmer.name.asc())
    )
    for name, kursname, standort in session.execute(stmt).all():
        print(name, "->", kursname, standort)

Praxisaufgabe (Mini)

Sie bauen eine kleine Kurs-DB (SQLite) und werten sie aus.

Beitrag zu den Lehr-/Lernzielen: LZ1 (Engine/Session), LZ2 (Modelle), LZ3 (CRUD), LZ4 (Queries).

  1. (LZ2) Definieren Sie Modelle Kurs und Teilnehmer (Relationship 1:n).
  2. (LZ2) Legen Sie die Tabellen an (Base.metadata.create_all).
  3. (LZ3) Fügen Sie 2 Kurse + je 2 Teilnehmer ein und committen Sie.
  4. (LZ4) Query: Alle Online-Kurse sortiert nach Preis; zusätzlich Ausgabe „Standort + Teilnehmeranzahl“.
  5. (Bonus) Aggregation: Durchschnittspreis pro Format (Online/Praesenz).
Lösungsvorschlag anzeigen
Lösung: Modelle + Insert + Queries (Beispieldaten) (Python)
from pathlib import Path
from sqlalchemy import create_engine, select, Integer, String, Float, ForeignKey, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker

Path("daten").mkdir(exist_ok=True)

class Base(DeclarativeBase):
    pass

class Kurs(Base):
    __tablename__ = "kurs"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    kursname: Mapped[str] = mapped_column(String(120), nullable=False)
    format: Mapped[str] = mapped_column(String(30), nullable=False)
    standort: Mapped[str] = mapped_column(String(60), nullable=False)
    dauer_tage: Mapped[int] = mapped_column(Integer, nullable=False)
    preis_euro: Mapped[float] = mapped_column(Float, nullable=False)

    teilnehmer: Mapped[list["Teilnehmer"]] = relationship(back_populates="kurs", cascade="all, delete-orphan")

class Teilnehmer(Base):
    __tablename__ = "teilnehmer"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(120), nullable=False)
    firma: Mapped[str] = mapped_column(String(120), nullable=False)

    kurs_id: Mapped[int] = mapped_column(ForeignKey("kurs.id"), nullable=False)
    kurs: Mapped["Kurs"] = relationship(back_populates="teilnehmer")

engine = create_engine("sqlite:///daten/sematrain.db", echo=False)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

with Session() as session:
    kurs1 = Kurs(kursname="Python Schulung", format="Online", standort="Berlin", dauer_tage=3, preis_euro=1390.00)
    kurs1.teilnehmer += [
        Teilnehmer(name="Max Beispiel", firma="Beispiel GmbH"),
        Teilnehmer(name="Ada Muster", firma="Muster AG")
    ]

    kurs2 = Kurs(kursname="Python Schulung", format="Praesenz", standort="Hamburg", dauer_tage=3, preis_euro=1490.00)
    kurs2.teilnehmer += [
        Teilnehmer(name="Lena Demo", firma="Demo KG"),
        Teilnehmer(name="Tim Test", firma="Test GmbH")
    ]

    session.add_all([kurs1, kurs2])
    session.commit()

with Session() as session:
    print("Online-Kurse (Preis aufsteigend):")
    stmt = select(Kurs).where(Kurs.format == "Online").order_by(Kurs.preis_euro.asc())
    for k in session.execute(stmt).scalars():
        print(k.standort, k.preis_euro, "TN:", len(k.teilnehmer))

with Session() as session:
    print("Ø Preis pro Format:")
    stmt = select(Kurs.format, func.avg(Kurs.preis_euro)).group_by(Kurs.format)
    for format_, avg_preis in session.execute(stmt).all():
        print(format_, round(avg_preis, 2))

Kurz-Takeaways

Quiz: Datenbanken mit SQLAlchemy

1. (LZ1) Wofür ist die SQLAlchemy Session in erster Linie da?

2. (LZ2) Wie heißen in SQLAlchemy ORM typischerweise die Klassen, die Tabellen abbilden?

3. (LZ3) Was passiert ohne commit() nach session.add(objekt)?

4. (LZ4) Welche Bausteine gehören typischerweise zu einer Query mit SQLAlchemy 2.x?

Praxisaufgabe

Mini-Projekt: Kurs- & Buchungsverwaltung (SQLite) mit SQLAlchemy ORM

Sie bauen eine kleine Datenbank im SemaTrain-Kontext (Beispieldaten): Kurse, Teilnehmer und Buchungen. Das Projekt ist absichtlich „realistisch“: Many-to-Many (TN ↔ Kurs) über eine Buchungstabelle, Rabattregel + Auswertungen.

Beitrag zu den Lehr-/Lernzielen:
LZ1 (Engine/Session), LZ2 (Modelle + Relationships), LZ3 (CRUD + commit/rollback), LZ4 (Joins + Aggregation/Reporting).

Aufgabe

Lösung anzeigen
Lösung: ORM-Setup + Buchungen (M:N) + CRUD + Report + JSON (Python)
from __future__ import annotations

import json
from datetime import datetime
from pathlib import Path

from sqlalchemy import (
    create_engine, select, func, ForeignKey,
    String, Integer, Float, DateTime, case
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker
)

# ===== LZ1: Setup (SQLite) =====
Path("daten").mkdir(exist_ok=True)
engine = create_engine("sqlite:///daten/sematrain.db", echo=False, future=True)
Session = sessionmaker(bind=engine, future=True)

# ===== LZ2: Modelle + Relationships =====
class Base(DeclarativeBase):
    pass

class Kurs(Base):
    __tablename__ = "kurs"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    kursname: Mapped[str] = mapped_column(String(120), nullable=False)
    format: Mapped[str] = mapped_column(String(30), nullable=False)     # Online / Praesenz
    standort: Mapped[str] = mapped_column(String(60), nullable=False)
    dauer_tage: Mapped[int] = mapped_column(Integer, nullable=False)
    preis_euro: Mapped[float] = mapped_column(Float, nullable=False)

    buchungen: Mapped[list["Buchung"]] = relationship(
        back_populates="kurs",
        cascade="all, delete-orphan"
    )

class Teilnehmer(Base):
    __tablename__ = "teilnehmer"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(120), nullable=False)
    firma: Mapped[str] = mapped_column(String(120), nullable=False)

    buchungen: Mapped[list["Buchung"]] = relationship(
        back_populates="teilnehmer",
        cascade="all, delete-orphan"
    )

class Buchung(Base):
    __tablename__ = "buchung"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)

    teilnehmer_id: Mapped[int] = mapped_column(ForeignKey("teilnehmer.id"), nullable=False)
    kurs_id: Mapped[int] = mapped_column(ForeignKey("kurs.id"), nullable=False)

    gebucht_am: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
    rabatt_prozent: Mapped[int] = mapped_column(Integer, default=0, nullable=False)

    teilnehmer: Mapped["Teilnehmer"] = relationship(back_populates="buchungen")
    kurs: Mapped["Kurs"] = relationship(back_populates="buchungen")

Base.metadata.create_all(engine)

# ===== Helper: Endpreis als SQL-Expression (LZ4) =====
def endpreis_expr():
    # Endpreis = preis * (1 - rabatt/100)
    return (Kurs.preis_euro * (1 - (Buchung.rabatt_prozent / 100.0)))

# ===== LZ3: Seed nur wenn leer =====
with Session() as session:
    hat_kurse = session.execute(select(func.count(Kurs.id))).scalar_one() > 0
    if not hat_kurse:
        kurs_hh = Kurs(kursname="Python Schulung", format="Praesenz", standort="Hamburg", dauer_tage=3, preis_euro=1490.00)
        kurs_be = Kurs(kursname="Python Schulung", format="Online",   standort="Berlin",  dauer_tage=3, preis_euro=1390.00)
        kurs_m  = Kurs(kursname="Python Schulung", format="Praesenz", standort="Muenchen",dauer_tage=4, preis_euro=1890.00)

        t1 = Teilnehmer(name="Max Beispiel", firma="Beispiel GmbH")
        t2 = Teilnehmer(name="Ada Muster",   firma="Muster AG")
        t3 = Teilnehmer(name="Lena Demo",    firma="Demo KG")

        session.add_all([kurs_hh, kurs_be, kurs_m, t1, t2, t3])
        session.flush()  # IDs verfügbar

        # 4 Buchungen (Beispieldaten)
        session.add_all([
            Buchung(teilnehmer=t1, kurs=kurs_be, rabatt_prozent=10),
            Buchung(teilnehmer=t2, kurs=kurs_be, rabatt_prozent=0),
            Buchung(teilnehmer=t3, kurs=kurs_hh, rabatt_prozent=5),
            Buchung(teilnehmer=t1, kurs=kurs_m,  rabatt_prozent=0),
        ])

        session.commit()
        print("Seed: Daten angelegt.")
    else:
        print("Seed: DB bereits gefüllt – übersprungen.")

# ===== LZ3: CRUD (Update/Delete/Create) + Fehlerbehandlung =====
try:
    with Session() as session:
        # UPDATE: Rabatt für eine Buchung setzen (z.B. erste Buchung von Max)
        b = session.execute(
            select(Buchung).join(Buchung.teilnehmer).where(Teilnehmer.name == "Max Beispiel")
        ).scalars().first()

        if b:
            b.rabatt_prozent = 15  # Beispiel
            session.commit()
            print("UPDATE: Rabatt gesetzt ->", b.rabatt_prozent, "%")

    with Session() as session:
        # DELETE: eine Buchung löschen (z.B. eine von Ada)
        b = session.execute(
            select(Buchung).join(Buchung.teilnehmer).where(Teilnehmer.name == "Ada Muster")
        ).scalars().first()

        if b:
            session.delete(b)
            session.commit()
            print("DELETE: Buchung von Ada gelöscht.")

    with Session() as session:
        # CREATE: neue Buchung anlegen (Lena bucht Berlin Online)
        lena = session.execute(select(Teilnehmer).where(Teilnehmer.name == "Lena Demo")).scalars().first()
        berlin = session.execute(select(Kurs).where(Kurs.standort == "Berlin", Kurs.format == "Online")).scalars().first()

        if lena and berlin:
            session.add(Buchung(teilnehmer=lena, kurs=berlin, rabatt_prozent=10))
            session.commit()
            print("CREATE: Neue Buchung (Lena -> Berlin Online).")

except Exception as e:
    # Wenn du Session außerhalb 'with' nutzt: session.rollback()
    print("FEHLER: Transaktion abgebrochen:", e)

# ===== LZ4: Query (Join) – Buchungsliste mit Endpreis =====
buchungsliste = []
with Session() as session:
    stmt = (
        select(
            Buchung.id,
            Teilnehmer.name,
            Teilnehmer.firma,
            Kurs.kursname,
            Kurs.format,
            Kurs.standort,
            Kurs.preis_euro,
            Buchung.rabatt_prozent,
            endpreis_expr().label("endpreis_euro"),
            Buchung.gebucht_am,
        )
        .join(Buchung.teilnehmer)
        .join(Buchung.kurs)
        .order_by(Kurs.standort.asc(), Teilnehmer.name.asc())
    )

    rows = session.execute(stmt).all()
    print("\nBuchungen:")
    for r in rows:
        print(
            f"#{r.id} | {r.name} ({r.firma}) | {r.kursname} | {r.format} | {r.standort} | "
            f"Preis: {r.preis_euro:.2f} | Rabatt: {r.rabatt_prozent}% | Endpreis: {r.endpreis_euro:.2f}"
        )
        buchungsliste.append({
            "buchung_id": r.id,
            "teilnehmer": r.name,
            "firma": r.firma,
            "kurs": r.kursname,
            "format": r.format,
            "standort": r.standort,
            "preis_euro": round(r.preis_euro, 2),
            "rabatt_prozent": r.rabatt_prozent,
            "endpreis_euro": round(float(r.endpreis_euro), 2),
            "gebucht_am": r.gebucht_am.isoformat(),
        })

# ===== Bonus (LZ4): Report – Umsatz (Beispiel) pro Standort =====
with Session() as session:
    stmt = (
        select(
            Kurs.standort,
            func.count(Buchung.id).label("buchungen"),
            func.sum(endpreis_expr()).label("umsatz_beispiel")
        )
        .join(Buchung.kurs)
        .group_by(Kurs.standort)
        .order_by(func.sum(endpreis_expr()).desc())
    )
    print("\nReport: Umsatz (Beispiel) je Standort")
    for standort, anzahl, umsatz in session.execute(stmt).all():
        print(standort, "| Buchungen:", anzahl, "| Umsatz:", round(float(umsatz or 0), 2))

# ===== Bonus: JSON Export (passt zu Kapitel 04) =====
export = Path("daten") / "buchungen_export.json"
export.write_text(json.dumps(buchungsliste, ensure_ascii=False, indent=2), encoding="utf-8")
print("\nJSON exportiert:", export)