Python-Kurs / API-Daten validieren und ohne doppelte IDs in SQLite sammeln

Prüfen Sie eine vollständige API-Antwort, bevor Sie ihre Datensätze anhand stabiler externer Kennungen speichern. Dieses Projekt verbindet eine JSON-Anfrage mit einer SQLite-Transaktion und aktualisiert bei erneut auftretender Kennung Name und Betrag, statt dieselbe Ausgabe nochmals anzulegen.

Identität, Aktualisierung und Vollständigkeit festlegen

Eine externe Kennung muss laut Schnittstellenvertrag dauerhaft denselben Vorgang identifizieren. Zwei Ausgaben mit gleichem Namen und Betrag können trotzdem unterschiedliche Vorgänge sein. Umgekehrt kann ein bereits bekannter Vorgang später einen korrigierten Betrag liefern. Das Beispiel verwendet deshalb id als Schlüssel und führt einen Upsert aus: neu einfügen oder den vorhandenen Datensatz aktualisieren.

Kommt eine Kennung innerhalb einer Antwort mehrfach vor, gewinnt hier der zuletzt verarbeitete Wert. Dies unterscheidet sich bewusst von einer Regel, die doppelte Kennungen als Fehler verwirft. Außerdem prüft das Programm key.strip() nur auf nichtleeren Inhalt; gespeichert wird die ursprüngliche Zeichenfolge. e1 und e1 sind daher unterschiedliche Schlüssel. Wenn ein Dienst etwas anderes verlangt, müssen Sie die Normalisierung vor dem Speichern ausdrücklich festlegen.

Alle Datensätze werden geprüft und in records gesammelt, bevor die Datenbank geöffnet wird. Ein unzulässiger Wert verhindert dadurch den gesamten Beginn des Speicherschritts. Die Datenbankoperationen selbst laufen in einer Transaktion. Die Netzwerkanfrage und die lokale Datenbank werden damit jedoch nicht zu einer gemeinsamen verteilten Transaktion.

Voraussetzungen sind Python 3.12 oder neuer und die Standardbibliothek. Die Grundlagen erläutern die HTTP-Client-Übung und der Kurs zu SQLite-Transaktionen.

Eine lokale Testantwort bereitstellen

Erstellen Sie einen neuen Übungsordner, getrennt von bisherigen Datenbanken und vertraulichen Dateien. Speichern Sie dort expenses.json als UTF-8:

[{"id":"e1","name":"coffee","amount":3500},{"id":"e2","name":"lunch","amount":8000}]

Starten Sie im ersten Terminal aus diesem Ordner den Testserver:

python -m http.server 8000 --bind 127.0.0.1

Die Bindung an 127.0.0.1 beschränkt den Testdienst auf diesen Rechner. Halten Sie das Terminal während der Client-Aufrufe offen. Falls Port 8000 bereits verwendet wird, wählen Sie einen anderen freien Port und ändern Sie die Client-URL entsprechend. Verwenden Sie für diese Übung keine Bindung an alle Netzwerkschnittstellen.

Der Dateiserver dient nur der kontrollierten Übung. Er stellt Dateien aus seinem Arbeitsordner bereit und ist kein produktiver API-Dienst. Öffnen Sie ein zweites Terminal mit derselben Projektumgebung. Unter macOS oder Ubuntu kann außerhalb einer virtuellen Umgebung der Interpreterbefehl python3 heißen.

Den vollständigen Collector ausführen

Speichern Sie im Übungsordner die vollständige Datei collect.py:

import json
import sqlite3
import sys
from urllib.request import urlopen

with urlopen(sys.argv[1], timeout=5) as response:
    if response.status != 200:
        raise ValueError("unexpected status")
    raw = response.read(1_000_001)
    if len(raw) > 1_000_000:
        raise ValueError("response too large")
    data = json.loads(raw.decode("utf-8"))
if not isinstance(data, list) or len(data) > 10000:
    raise ValueError("invalid batch")
MAX_AMOUNT = 1_000_000_000_000
records = []
for row in data:
    if not isinstance(row, dict):
        raise ValueError("invalid row")
    key, name, amount = row.get("id"), row.get("name"), row.get("amount")
    if not isinstance(key, str) or not key.strip() or not isinstance(name, str) or not name.strip():
        raise ValueError("invalid identity")
    if type(amount) is not int or not 0 <= amount <= MAX_AMOUNT:
        raise ValueError("invalid amount")
    records.append((key, name, amount))
con = sqlite3.connect("collected.db", autocommit=False)
try:
    with con:
        con.execute("CREATE TABLE IF NOT EXISTS expenses (id TEXT PRIMARY KEY, name TEXT NOT NULL, amount INTEGER NOT NULL CHECK(amount BETWEEN 0 AND 1000000000000))")
        con.executemany("INSERT INTO expenses VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, amount=excluded.amount", records)
    count = con.execute("SELECT COUNT(*) FROM expenses").fetchone()[0]
    total = sum(row[0] for row in con.execute("SELECT amount FROM expenses"))
    print("stored:", count, "total:", total)
finally:
    con.close()

Starten Sie im zweiten Terminal:

python collect.py http://127.0.0.1:8000/expenses.json

Beim ersten erfolgreichen Lauf erscheinen stored: 2 total: 11500 und die Datenbankdatei collected.db. Die Anzahl und Summe beziehen sich auf den gesamten aktuellen Datenbankbestand, nicht allein auf die gerade empfangene Antwort. Bei identischer Wiederholung müssen es weiterhin zwei Datensätze und 11.500 bleiben. Wenn Sie nur den Betrag von e2 auf 9000 ändern, erwarten Sie zwei Datensätze und 12.500.

Vor der Verarbeitung prüft der Client Status 200, maximal 1.000.000 Antwortbytes, UTF-8 und JSON. Die äußere Liste darf höchstens 10.000 Elemente besitzen. Jeder Eintrag muss ein Dictionary mit nichtleerer Zeichenfolgen-Kennung und nichtleerem Namen sein. Beträge sind echte Ganzzahlen von null bis 1.000.000.000.000; boolesche Werte und Zahlen als Text werden abgelehnt.

Eine Ausnahme führt in dieser kompakten Übung zum Traceback und Fehlerstatus. Der Aufruf benötigt die URL als Parameter; ein fehlender Parameter wird hier noch nicht mit einer komfortablen argparse-Hilfe behandelt. Werten Sie bei einem Fehler den Status und die Ursache aus, bevor Sie abhängige Schritte starten. Ersetzen Sie eine fehlgeschlagene Anfrage nicht durch eine vermeintlich erfolgreiche leere Liste.

executemany verwendet die vorbereiteten Werte als SQL-Parameter. Der Kontextmanager bestätigt die Schreiboperationen nur bei erfolgreichem Abschluss und rollt sie bei einer nach außen laufenden Ausnahme zurück. Das finally schließt die Verbindung separat. Die Referenz zu sqlite3 beschreibt diese Operationen.

Für die Gesamtsumme liest das Beispiel die gespeicherten Ganzzahlen und addiert sie mit Python sum. SQLite SUM kann bei ganzzahligem Ergebnis einen Überlauf melden, wenn eine sehr große Gesamtsumme die 64-Bit-Grenze überschreitet. Der Bestand kann über viele Läufe wachsen, auch wenn die Einzelantwort begrenzt ist. Ein Wechsel auf das Gleitkomma-Ergebnis von total() würde die Anforderung exakter Ganzzahlbeträge verändern. Die SQLite-Referenz zu sum und total erläutert den Unterschied.

Wiederholung, Fehler und fehlende Datensätze kontrollieren

1. Verarbeiten Sie die ursprüngliche Antwort zweimal und vergleichen Sie Anzahl und Summe. Beide Läufe müssen zwei Zeilen mit insgesamt 11.500 ergeben.
2. Ändern Sie ausschließlich den Betrag von e2 auf 9000. Prüfen Sie zwei gespeicherte Datensätze mit insgesamt 12.500.
3. Ändern Sie danach einen Betrag in die JSON-Zeichenfolge "9000". Die Eingabe muss vor dem Öffnen der Datenbank abgelehnt werden; der vorherige Inhalt muss erhalten bleiben. Stellen Sie anschließend gültige Testdaten wieder her.
4. Beenden Sie den Testserver nach der Übung mit Ctrl+C im ersten Terminal.

Ein Datensatz, der in einer späteren Antwort fehlt, wird nicht gelöscht. Auch eine gültige leere Liste leert die bestehende Datenbank nicht; die nachfolgende Ausgabe zeigt weiterhin deren Inhalt. Der Collector ist somit kein vollständiger Abgleich eines entfernten Bestands. Legen Sie vor einer Erweiterung fest, ob fehlende Daten als unverändert, gelöscht oder noch nicht geladen gelten.

Bei externen Diensten müssen Sie außerdem Seitennavigation, Authentifizierung, Anfragelimits und Änderungen zwischen Seiten berücksichtigen. Der Erfolg einer einzelnen Seite beweist keine vollständig abgeschlossene Sammlung. timeout=5 begrenzt blockierende Netzwerkoperationen, ist aber keine feste Gesamtlaufzeit. Die Referenz zu urlopen beschreibt die Anfrageschnittstelle.

Die Datenbankdefinition wird bei vorhandener Tabelle nicht automatisch erneuert. Nutzen Sie die neue Übungsdatei oder eine geplante Migration mit Sicherung für reale Bestände. Ebenso verhindert die ursprüngliche INTEGER-Spaltendeklaration nicht jede fremde unpassende Schreiboperation; der Collector verlässt sich zusätzlich auf seine eigene Vorprüfung. Begrenzen Sie Wiederholungen und behalten Sie Fehler sichtbar, statt Datenlücken durch endlose Versuche oder leere Ersatzantworten zu verdecken.