SPF, DKIM und DMARC: E-Mail-Versand vertrauenswürdig machen

SQLAlchemy ist die verbreitetste Bibliothek für Datenbankzugriff in Python. Sie bietet zwei Ebenen: eine Core genannte Schicht für SQL-nahe Arbeit und einen ORM-Teil, der Tabellen auf Python-Klassen abbildet. Dieser Artikel zeigt beide und wann welche sinnvoll ist.

Die zwei Ebenen

Diese Unterscheidung ist der Schlüssel zum Verständnis:

Core: Man arbeitet mit Tabellen und Ausdrücken, die zu SQL werden. Das Ergebnis sind Zeilen, keine Objekte. Nahe am SQL, ohne Verstecken.
ORM (Object Relational Mapper): Python-Klassen mit Attributen; Zugriffe auf Beziehungen werden automatisch zu Verknüpfungen. Bequem für Anwendungslogik, riskant bei unüberlegter Verwendung.

Der größte Vorteil beider Ebenen: parametrisierte Abfragen. Werte werden nie in SQL-Zeichenketten eingesetzt, sondern als Parameter übergeben. Damit ist SQL-Injection strukturell ausgeschlossen.

Verbindung und Modell

# requirements: sqlalchemy>=2.0   psycopg[binary]  oder  pymysql

from datetime import datetime
from sqlalchemy import create_engine, String, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class Team(Base):
    __tablename__ = "teams"
    id:   Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(80), unique=True, index=True)
    mitglieder: Mapped[list["Server"]] = relationship(back_populates="team")

class Server(Base):
    __tablename__ = "server"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str] = mapped_column(String(80), index=True)
    ip:    Mapped[str] = mapped_column(String(45))       # 45 = IPv6-Laenge
    port:  Mapped[int] = mapped_column(default=22)
    aktiv: Mapped[bool] = mapped_column(default=True)
    angelegt: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now())
    team_id: Mapped[int | None] = mapped_column(ForeignKey("teams.id"))
    team: Mapped[Team | None] = relationship(back_populates="mitglieder")

# Engine: Verbindungspool je Anwendung, nicht je Abfrage
engine = create_engine(
    "postgresql+psycopg://benutzer:Diese E-Mail-Adresse ist vor Spambots geschützt! Zur Anzeige muss JavaScript eingeschaltet sein./appdb",
    pool_size=5, max_overflow=10, pool_pre_ping=True, echo=False)

Base.metadata.create_all(engine)      # Tabellen anlegen (nur zum Start)

pool_pre_ping=True ist im Betrieb wichtig: Vor jeder Ausleihe aus dem Verbindungspool wird geprüft, ob die Verbindung noch lebt. Ohne das schlägt die erste Abfrage nach einer Netzwerkunterbrechung mit einem Fehler fehl.

Abfragen mit dem ORM

from sqlalchemy import select, func, and_, or_

with Session(engine) as s:
    # Einfuegen
    team = Team(name="Betrieb")
    s.add(team)
    s.flush()                          # team.id ist jetzt bekannt
    s.add_all([
        Server(name="web01", ip="10.0.0.11", team=team),
        Server(name="web02", ip="10.0.0.12", team=team),
    ])
    s.commit()

    # Lesen (select statt query in SQLAlchemy 2.0)
    alle = s.scalars(select(Server)).all()

    # Filtern - IMMER mit Parametern, nie mit f-Strings
    aktiv = s.scalars(
        select(Server).where(Server.aktiv.is_(True)).order_by(Server.name)
    ).all()

    # Mehrere Bedingungen
    treffer = s.scalars(
        select(Server).where(
            and_(Server.aktiv.is_(True), Server.port != 22)
        )
    ).all()

    # Einzelnes Ergebnis
    einer = s.scalar(select(Server).where(Server.name == "web01"))
    #        ^-- scalar: genau eines oder None
    #            scalars().first(): erstes oder None

    # Zaehlen, Summe, Gruppierung
    anzahl = s.scalar(select(func.count()).select_from(Server))
    je_team = s.execute(
        select(Team.name, func.count(Server.id))
        .join(Server, Server.team_id == Team.id)
        .group_by(Team.name)
    ).all()

    # Aendern
    einer.port = 2222
    s.commit()                          # Aenderung wird geschrieben

    # Loeschen
    s.delete(einer)
    s.commit()

Transaktionen

with Session(engine) as s:
    try:
        s.add(Server(name="neu", ip="10.0.0.99"))
        s.flush()                       # schickt SQL, noch kein commit
        s.commit()                      # erst hier dauerhaft
    except Exception:
        s.rollback()
        raise

# Mehrere Vorgaenge in EINER Transaktion
with Session(engine) as s:
    with s.begin():                     # commit am Ende, rollback bei Fehler
        s.add(Server(name="a", ip="10.0.0.1"))
        s.add(Server(name="b", ip="10.0.0.2"))
        # Bricht hier etwas ab, wird KEINE der beiden Zeilen geschrieben.

# Wichtig: Eine Sitzung ist kein Ersatz fuer eine Transaktion ueber
# mehrere Anfragen. Eine Sitzung lebt so lange wie die Anfrage,
# nicht so lange wie eine Benutzersitzung.

Die N+1-Falle

Das ist das häufigste Leistungsproblem mit ORMs. Ein naiver Zugriff auf eine Beziehung erzeugt eine Abfrage je Datensatz:

# SCHLECHT: 1 Abfrage fuer alle Server, dann 1 JE Server fuer das Team
server = s.scalars(select(Server)).all()
for srv in server:
    print(srv.name, srv.team.name)      # 100 Server = 101 Abfragen!

# GUT: alles in einer Abfrage laden
from sqlalchemy.orm import selectinload, joinedload

server = s.scalars(
    select(Server).options(selectinload(Server.team))
).all()
# 2 Abfragen insgesamt, unabhaengig von der Anzahl

# Alternative: joinedload macht eine Abfrage mit JOIN
#   selectinload: 2 Abfragen, saubere Trennung, gut bei 1:n
#   joinedload:   1 Abfrage mit JOIN, gut bei n:1
# Beides laesst sich verschachteln: .options(selectinload(A.b).joinedload(B.c))

# Gegenprobe: Alle Abfragen mitloggen
engine = create_engine(url, echo=True)      # zeigt jedes Statement
# In Produktion besser: slow-query-Log der Datenbank auswerten.

Core: SQL-nah arbeiten

Für Massenoperationen, Berichte und Aufräumläufe ist Core oft die bessere Wahl – vorhersehbarer und schneller.

from sqlalchemy import text, insert, update, delete, bindparam

with engine.begin() as conn:
    # Masseneinfuegen: deutlich schneller als 1000 Einzelabfragen
    conn.execute(
        insert(Server),
        [{"name": f"node{i:03d}", "ip": f"10.0.1.{i}"} for i in range(1, 1001)]
    )

    # Massenupdate
    conn.execute(
        update(Server).where(Server.ip.like("10.0.1.%")).values(aktiv=False)
    )

    # Massenloeschen
    conn.execute(delete(Server).where(Server.aktiv.is_(False)))

    # Rohes SQL - aber MIT Parametern!
    rows = conn.execute(
        text("SELECT name, ip FROM server WHERE port > :p ORDER BY name LIMIT :n"),
        {"p": 1024, "n": 10}
    ).all()

    # Ergebnis als Woerter
    for row in rows:
        print(row._mapping["name"], row.ip)

Wann ORM, wann Core

ORM: Anwendungslogik, einzelne Datensätze, Beziehungen, Validierung im Modell. Der Normalfall für Webanwendungen.
Core: Massenoperationen, Berichte mit mehreren Verknüpfungen, Datenmigrationen, Aufräumläufe. Alles, wo die Anzahl der Zeilen groß ist und keine Objektlogik gebraucht wird.
Rohes SQL: wenn die Abfrage mit den Mitteln der Bibliothek nicht ausdrückbar ist – Fensterfunktionen, spezifische Erweiterungen. Dann aber immer über text() mit gebundenen Parametern.

Migrationen

Base.metadata.create_all() legt fehlende Tabellen an, ändert aber nichts an bestehenden. Für Änderungen an einer laufenden Datenbank braucht es ein Migrationswerkzeug – Alembic ist der Standard und arbeitet mit SQLAlchemy zusammen:

alembic init migrationen
# alembic.ini und env.py anpassen (Datenbankadresse + Base.metadata)

alembic revision --autogenerate -m "spalte letzte_pruefung ergaenzt"
#   Datei PRUEFEN! Die automatische Erkennung uebersieht Aenderungen
#   und schlaegt manchmal Loeschungen vor, die nicht gewollt sind.

alembic upgrade head        # anwenden
alembic downgrade -1        # eine Stufe zurueck
alembic history
alembic current

Wichtig: Auch Alembic erkennt nicht alles. Umbenennungen von Spalten werden oft als Löschen-und-Neuanlegen interpretiert – mit Datenverlust. Die erzeugte Migrationsdatei ist deshalb immer zu lesen und bei Bedarf von Hand zu korrigieren.

Prüfliste für den Betrieb

Verbindungspool richtig dimensionieren – nicht pro Anfrage eine neue Verbindung.
pool_pre_ping=True und ein pool_recycle unterhalb der Zeitgrenze der Datenbank.
Sitzungen kurz halten und nach jeder Anfrage schließen (with Session(engine)).
Beziehungen mit selectinload oder joinedload laden, wenn sie gebraucht werden.
Massenoperationen über Core, nicht über ORM-Objekte.
Zugangsdaten aus der Umgebung, nie im Quelltext.
Indizes im Modell deklarieren (index=True) – und nach dem Anlegen mit sqlalchemy.explain oder dem Datenbankplan gegenprüfen.
Migrationen versioniert im Repository, erzeugte Dateien vor dem Anwenden geprüft.

Netzwerk: DNS & Dienste