Abschnitt 17 · Lektion 4
Synthese: Abholen und speichern
Ein Programm, das jeden Morgen die Vorhersage holt und als Tabelle ablegt, ist die Sorte Werkzeug, für die man Python lernt. Bis dahin fehlen noch zwei Dinge. Echte Dienste wollen fast immer wissen, wer fragt, und dafür gibt es Schlüssel. Und was abgeholt ist, soll bleiben. Also kommt es in eine Datei.
Ein Schlüssel für den Dienst
# Der Wetterdienst für diesen Abschnitt. Die Maschine hier hat kein Netz, deshalb
# läuft der Dienst auf derselben Maschine wie dein Programm, unter der Adresse
# http://127.0.0.1:8000. Die Anfragen an ihn sind trotzdem echt: Sie gehen über
# HTTP hinaus und kommen als Antwort zurück, nur eben nicht durch das Internet.
# Hier musst du nichts ändern.
import hashlib
import json
import threading
from datetime import date, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
# Das Wetter am 09.10.2026 um 14 Uhr: Grad, Wetter, Wind in km/h.
AKTUELL = {
"Berlin": (14.2, "bewölkt", 12),
"Hamburg": (12.8, "Regen", 24),
"München": (9.6, "Nebel", 3),
"Köln": (15.1, "heiter", 8),
"Frankfurt am Main": (13.4, "bewölkt", 10),
"Stuttgart": (11.9, "Regen", 14),
"Hannover": (13.1, "bewölkt", 16),
"Nürnberg": (10.8, "Nebel", 4),
"Leipzig": (13.7, "heiter", 9),
"Dresden": (12.3, "bewölkt", 11),
"Erfurt": (12.6, "heiter", 7),
}
# Die Vorhersage ab dem 09.10.2026, je Tag: Höchstwert, Tiefstwert, Regen in mm
# und das Wetter.
VORHERSAGE = {
"Leipzig": [
(15.2, 6.1, 0.0, "heiter"),
(17.8, 7.4, 0.0, "sonnig"),
(18.4, 8.0, 0.0, "sonnig"),
(16.9, 9.2, 0.4, "bewölkt"),
(13.5, 8.8, 6.2, "Regen"),
(11.2, 7.5, 9.8, "Regen"),
(12.6, 5.9, 1.1, "bewölkt"),
],
"Dresden": [
(14.6, 5.2, 0.2, "bewölkt"),
(16.3, 6.8, 0.0, "heiter"),
(15.0, 8.1, 0.0, "bewölkt"),
(16.3, 7.7, 0.0, "heiter"),
(19.1, 8.4, 0.0, "sonnig"),
(12.4, 7.0, 5.1, "Regen"),
(10.8, 6.2, 7.6, "Regen"),
],
"Erfurt": [
(13.9, 4.8, 0.0, "heiter"),
(15.6, 5.5, 0.0, "sonnig"),
(14.1, 7.2, 2.3, "Regen"),
(12.0, 6.6, 4.0, "Regen"),
(12.8, 5.1, 0.6, "bewölkt"),
(14.7, 4.3, 0.0, "heiter"),
(13.3, 3.9, 0.0, "heiter"),
],
}
# Für diese Städte klappt etwas nicht, so wie es bei echten Diensten vorkommt.
STOERUNGEN = {
"Bremen": 500, # der Dienst hat selbst einen Fehler
"Kiel": 503, # der Dienst wird gerade gewartet
"Rostock": "keine Antwort", # er nimmt die Anfrage an und antwortet nie
"Lübeck": "Abbruch", # er legt auf, ohne zu antworten
}
# Die Vorhersage gibt es nur mit Schlüssel in der Kopfzeile X-Api-Key. Wie jeder
# ordentliche Dienst kennt dieser den Schlüssel nicht selbst, sondern nur eine
# Prüfsumme davon.
PRUEFSUMME = "0fe6daa0a6754b2239b7022999ec7a5ae2b20fc096afc6d9eead51d07bef68ce"
class Anfrage(BaseHTTPRequestHandler):
def do_GET(self):
adresse = urlparse(self.path)
werte = {name: liste[0] for name, liste in parse_qs(adresse.query).items()}
stadt = werte.get("stadt", "")
stoerung = STOERUNGEN.get(stadt)
if adresse.path not in ("/wetter", "/vorhersage"):
self.sende(404, {"fehler": f"Unbekannte Adresse: {adresse.path}"})
elif stoerung == "keine Antwort":
# Wartet auf etwas, das nie passiert. Der Dienst endet trotzdem mit
# deinem Programm, weil er im Hintergrund läuft.
threading.Event().wait()
elif stoerung == "Abbruch":
self.close_connection = True
elif stoerung is not None:
self.sende(stoerung, self.responses[stoerung][0])
elif adresse.path == "/wetter":
self.wetter(stadt)
else:
self.vorhersage(stadt, werte.get("tage", "7"))
def wetter(self, stadt):
if stadt not in AKTUELL:
self.sende(404, {"fehler": f"Unbekannte Stadt: {stadt}"})
return
grad, wetter, wind = AKTUELL[stadt]
aktuell = {"temperatur": grad, "wetter": wetter, "wind": wind}
self.sende(200, {"stadt": stadt, "zeit": "2026-10-09T14:00", "aktuell": aktuell})
def vorhersage(self, stadt, tage):
schluessel = self.headers.get("X-Api-Key", "")
if hashlib.sha256(schluessel.encode()).hexdigest() != PRUEFSUMME:
self.sende(401, {"fehler": "Ohne gültigen Schlüssel in X-Api-Key gibt es keine Vorhersage"})
elif stadt not in VORHERSAGE:
self.sende(404, {"fehler": f"Keine Vorhersage für: {stadt}"})
elif not tage.isdigit() or not 1 <= int(tage) <= 7:
self.sende(400, {"fehler": "tage geht von 1 bis 7"})
else:
liste = []
for nummer, (hoch, tief, regen, wetter) in enumerate(VORHERSAGE[stadt][: int(tage)]):
liste.append(
{
"datum": (date(2026, 10, 9) + timedelta(days=nummer)).isoformat(),
"wetter": wetter,
"hoechstwert": hoch,
"tiefstwert": tief,
"regen": regen,
}
)
self.sende(200, {"stadt": stadt, "tage": liste})
def sende(self, status, inhalt):
"""Schickt die Antwort: ein Dictionary als JSON, einen Text als Text."""
if isinstance(inhalt, str):
art, rumpf = "text/plain", inhalt
else:
art, rumpf = "application/json", json.dumps(inhalt, ensure_ascii=False)
daten = rumpf.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", f"{art}; charset=utf-8")
self.send_header("Content-Length", str(len(daten)))
self.end_headers()
self.wfile.write(daten)
def log_message(self, *angaben):
# Sonst stünde jede Anfrage auf dem Fehlerkanal deines Programms.
pass
_server = None
def starten():
"""Startet den Dienst im Hintergrund. Er läuft, bis dein Programm endet."""
global _server
if _server is not None:
return
try:
_server = ThreadingHTTPServer(("127.0.0.1", 8000), Anfrage)
except OSError:
# Port 8000 ist belegt, weil der Dienst aus einem anderen Lauf noch
# antwortet. Dann fragt dein Programm eben den.
return
threading.Thread(target=_server.serve_forever, daemon=True).start() import requests
import wetterdienst
wetterdienst.starten()
ADRESSE = "http://127.0.0.1:8000"
antwort = requests.get(ADRESSE + "/vorhersage", params={"stadt": "Erfurt", "tage": 3}, timeout=2)
print(antwort.status_code, antwort.text)
antwort.raise_for_status()
print("Hier kommt das Programm nicht mehr hin.") Das aktuelle Wetter bekommt beim Wetterdienst jeder, die Vorhersage nicht. Ohne Schlüssel kommt
Status 401 zurück, und im Rumpf steht, was fehlt, nämlich ein gültiger Schlüssel in der Kopfzeile
X-Api-Key. Mit raise_for_status() wird daraus ein HTTPError, 401 Client Error: Unauthorized.
Das Programm endet also wie in der letzten Lektion an der richtigen Stelle, mit einer Meldung, in
der der Status steht.
So ein Schlüssel (API key) ist ein langer Text aus Buchstaben und Ziffern. Man bekommt ihn, wenn man sich beim Anbieter anmeldet, und am Schlüssel erkennt der Dienst, wer fragt. Davon hängt ab, wie oft man fragen darf und manchmal auch, was es kostet. Wer Unfug treibt, dem wird der Schlüssel gesperrt.
Meistens geht der Schlüssel in einer Kopfzeile der Anfrage mit, so wie hier. Manchmal steht er auch
als Parameter in der Adresse. Die Kopfzeile ist die bessere Wahl, weil eine Adresse an vielen
Stellen landet, etwa in den Protokollen des Dienstes oder in der Meldung von raise_for_status()
oben, in der die ganze Adresse ausgeschrieben ist.
Der Schlüssel gehört nicht in den Code
# Der Wetterdienst für diesen Abschnitt. Die Maschine hier hat kein Netz, deshalb
# läuft der Dienst auf derselben Maschine wie dein Programm, unter der Adresse
# http://127.0.0.1:8000. Die Anfragen an ihn sind trotzdem echt: Sie gehen über
# HTTP hinaus und kommen als Antwort zurück, nur eben nicht durch das Internet.
# Hier musst du nichts ändern.
import hashlib
import json
import threading
from datetime import date, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
# Das Wetter am 09.10.2026 um 14 Uhr: Grad, Wetter, Wind in km/h.
AKTUELL = {
"Berlin": (14.2, "bewölkt", 12),
"Hamburg": (12.8, "Regen", 24),
"München": (9.6, "Nebel", 3),
"Köln": (15.1, "heiter", 8),
"Frankfurt am Main": (13.4, "bewölkt", 10),
"Stuttgart": (11.9, "Regen", 14),
"Hannover": (13.1, "bewölkt", 16),
"Nürnberg": (10.8, "Nebel", 4),
"Leipzig": (13.7, "heiter", 9),
"Dresden": (12.3, "bewölkt", 11),
"Erfurt": (12.6, "heiter", 7),
}
# Die Vorhersage ab dem 09.10.2026, je Tag: Höchstwert, Tiefstwert, Regen in mm
# und das Wetter.
VORHERSAGE = {
"Leipzig": [
(15.2, 6.1, 0.0, "heiter"),
(17.8, 7.4, 0.0, "sonnig"),
(18.4, 8.0, 0.0, "sonnig"),
(16.9, 9.2, 0.4, "bewölkt"),
(13.5, 8.8, 6.2, "Regen"),
(11.2, 7.5, 9.8, "Regen"),
(12.6, 5.9, 1.1, "bewölkt"),
],
"Dresden": [
(14.6, 5.2, 0.2, "bewölkt"),
(16.3, 6.8, 0.0, "heiter"),
(15.0, 8.1, 0.0, "bewölkt"),
(16.3, 7.7, 0.0, "heiter"),
(19.1, 8.4, 0.0, "sonnig"),
(12.4, 7.0, 5.1, "Regen"),
(10.8, 6.2, 7.6, "Regen"),
],
"Erfurt": [
(13.9, 4.8, 0.0, "heiter"),
(15.6, 5.5, 0.0, "sonnig"),
(14.1, 7.2, 2.3, "Regen"),
(12.0, 6.6, 4.0, "Regen"),
(12.8, 5.1, 0.6, "bewölkt"),
(14.7, 4.3, 0.0, "heiter"),
(13.3, 3.9, 0.0, "heiter"),
],
}
# Für diese Städte klappt etwas nicht, so wie es bei echten Diensten vorkommt.
STOERUNGEN = {
"Bremen": 500, # der Dienst hat selbst einen Fehler
"Kiel": 503, # der Dienst wird gerade gewartet
"Rostock": "keine Antwort", # er nimmt die Anfrage an und antwortet nie
"Lübeck": "Abbruch", # er legt auf, ohne zu antworten
}
# Die Vorhersage gibt es nur mit Schlüssel in der Kopfzeile X-Api-Key. Wie jeder
# ordentliche Dienst kennt dieser den Schlüssel nicht selbst, sondern nur eine
# Prüfsumme davon.
PRUEFSUMME = "0fe6daa0a6754b2239b7022999ec7a5ae2b20fc096afc6d9eead51d07bef68ce"
class Anfrage(BaseHTTPRequestHandler):
def do_GET(self):
adresse = urlparse(self.path)
werte = {name: liste[0] for name, liste in parse_qs(adresse.query).items()}
stadt = werte.get("stadt", "")
stoerung = STOERUNGEN.get(stadt)
if adresse.path not in ("/wetter", "/vorhersage"):
self.sende(404, {"fehler": f"Unbekannte Adresse: {adresse.path}"})
elif stoerung == "keine Antwort":
# Wartet auf etwas, das nie passiert. Der Dienst endet trotzdem mit
# deinem Programm, weil er im Hintergrund läuft.
threading.Event().wait()
elif stoerung == "Abbruch":
self.close_connection = True
elif stoerung is not None:
self.sende(stoerung, self.responses[stoerung][0])
elif adresse.path == "/wetter":
self.wetter(stadt)
else:
self.vorhersage(stadt, werte.get("tage", "7"))
def wetter(self, stadt):
if stadt not in AKTUELL:
self.sende(404, {"fehler": f"Unbekannte Stadt: {stadt}"})
return
grad, wetter, wind = AKTUELL[stadt]
aktuell = {"temperatur": grad, "wetter": wetter, "wind": wind}
self.sende(200, {"stadt": stadt, "zeit": "2026-10-09T14:00", "aktuell": aktuell})
def vorhersage(self, stadt, tage):
schluessel = self.headers.get("X-Api-Key", "")
if hashlib.sha256(schluessel.encode()).hexdigest() != PRUEFSUMME:
self.sende(401, {"fehler": "Ohne gültigen Schlüssel in X-Api-Key gibt es keine Vorhersage"})
elif stadt not in VORHERSAGE:
self.sende(404, {"fehler": f"Keine Vorhersage für: {stadt}"})
elif not tage.isdigit() or not 1 <= int(tage) <= 7:
self.sende(400, {"fehler": "tage geht von 1 bis 7"})
else:
liste = []
for nummer, (hoch, tief, regen, wetter) in enumerate(VORHERSAGE[stadt][: int(tage)]):
liste.append(
{
"datum": (date(2026, 10, 9) + timedelta(days=nummer)).isoformat(),
"wetter": wetter,
"hoechstwert": hoch,
"tiefstwert": tief,
"regen": regen,
}
)
self.sende(200, {"stadt": stadt, "tage": liste})
def sende(self, status, inhalt):
"""Schickt die Antwort: ein Dictionary als JSON, einen Text als Text."""
if isinstance(inhalt, str):
art, rumpf = "text/plain", inhalt
else:
art, rumpf = "application/json", json.dumps(inhalt, ensure_ascii=False)
daten = rumpf.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", f"{art}; charset=utf-8")
self.send_header("Content-Length", str(len(daten)))
self.end_headers()
self.wfile.write(daten)
def log_message(self, *angaben):
# Sonst stünde jede Anfrage auf dem Fehlerkanal deines Programms.
pass
_server = None
def starten():
"""Startet den Dienst im Hintergrund. Er läuft, bis dein Programm endet."""
global _server
if _server is not None:
return
try:
_server = ThreadingHTTPServer(("127.0.0.1", 8000), Anfrage)
except OSError:
# Port 8000 ist belegt, weil der Dienst aus einem anderen Lauf noch
# antwortet. Dann fragt dein Programm eben den.
return
threading.Thread(target=_server.serve_forever, daemon=True).start() import os
import requests
import wetterdienst
wetterdienst.starten()
ADRESSE = "http://127.0.0.1:8000"
schluessel = os.environ["WETTER_SCHLUESSEL"]
antwort = requests.get(
ADRESSE + "/vorhersage",
params={"stadt": "Erfurt", "tage": 3},
headers={"X-Api-Key": schluessel},
timeout=2,
)
antwort.raise_for_status()
for tag in antwort.json()["tage"]:
print(tag["datum"], tag["wetter"], tag["hoechstwert"]) Am einfachsten wäre es, den Schlüssel direkt in main.py zu schreiben. Dann steckt er aber in jeder
Kopie der Datei: in der, die du einer Kollegin schickst, in der, die du in einem Forum zeigst, weil
etwas nicht klappt, und in jeder Sicherung. Wer ihn findet, fragt auf deine Rechnung. Sicherer ist
die Annahme, dass jede Datei mit Code irgendwann jemand anderes liest. Meistens stimmt das auch.
Der Schlüssel kommt deshalb aus einer Umgebungsvariablen (environment variable). Das sind Werte,
die das Betriebssystem einem Programm beim Start mitgibt, jeden unter einem Namen. In Python stehen
sie in os.environ, einem Dictionary aus dem Modul os, und os.environ["WETTER_SCHLUESSEL"] ist
der Wert unter diesem Namen. Im Code steht also nur der Name. Der Schlüssel selbst liegt auf dem
Rechner, auf dem das Programm läuft.
Wenn du dieses Beispiel mit Sandbox starten laufen lässt, ist die Variable auf der Maschine
schon gesetzt. Im Terminal ist sie es nicht, wie auf jedem Rechner, auf dem das noch niemand getan
hat. Dort gibt der erste Befehl der Befehlsleiste, WETTER_SCHLUESSEL=… python main.py, die
Variable genau diesem einen Befehl mit. Am zweiten, python main.py, siehst du, was ohne sie
passiert.
Die Kopfzeile geht genauso als Dictionary mit wie die Parameter:
headers={"X-Api-Key": schluessel}. Wie sie heißen muss, steht in der Beschreibung des Dienstes. Da
steht auch, ob er den Schlüssel überhaupt auf diesem Weg haben will.
Eine Variable, die fehlt
# Der Wetterdienst für diesen Abschnitt. Die Maschine hier hat kein Netz, deshalb
# läuft der Dienst auf derselben Maschine wie dein Programm, unter der Adresse
# http://127.0.0.1:8000. Die Anfragen an ihn sind trotzdem echt: Sie gehen über
# HTTP hinaus und kommen als Antwort zurück, nur eben nicht durch das Internet.
# Hier musst du nichts ändern.
import hashlib
import json
import threading
from datetime import date, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
# Das Wetter am 09.10.2026 um 14 Uhr: Grad, Wetter, Wind in km/h.
AKTUELL = {
"Berlin": (14.2, "bewölkt", 12),
"Hamburg": (12.8, "Regen", 24),
"München": (9.6, "Nebel", 3),
"Köln": (15.1, "heiter", 8),
"Frankfurt am Main": (13.4, "bewölkt", 10),
"Stuttgart": (11.9, "Regen", 14),
"Hannover": (13.1, "bewölkt", 16),
"Nürnberg": (10.8, "Nebel", 4),
"Leipzig": (13.7, "heiter", 9),
"Dresden": (12.3, "bewölkt", 11),
"Erfurt": (12.6, "heiter", 7),
}
# Die Vorhersage ab dem 09.10.2026, je Tag: Höchstwert, Tiefstwert, Regen in mm
# und das Wetter.
VORHERSAGE = {
"Leipzig": [
(15.2, 6.1, 0.0, "heiter"),
(17.8, 7.4, 0.0, "sonnig"),
(18.4, 8.0, 0.0, "sonnig"),
(16.9, 9.2, 0.4, "bewölkt"),
(13.5, 8.8, 6.2, "Regen"),
(11.2, 7.5, 9.8, "Regen"),
(12.6, 5.9, 1.1, "bewölkt"),
],
"Dresden": [
(14.6, 5.2, 0.2, "bewölkt"),
(16.3, 6.8, 0.0, "heiter"),
(15.0, 8.1, 0.0, "bewölkt"),
(16.3, 7.7, 0.0, "heiter"),
(19.1, 8.4, 0.0, "sonnig"),
(12.4, 7.0, 5.1, "Regen"),
(10.8, 6.2, 7.6, "Regen"),
],
"Erfurt": [
(13.9, 4.8, 0.0, "heiter"),
(15.6, 5.5, 0.0, "sonnig"),
(14.1, 7.2, 2.3, "Regen"),
(12.0, 6.6, 4.0, "Regen"),
(12.8, 5.1, 0.6, "bewölkt"),
(14.7, 4.3, 0.0, "heiter"),
(13.3, 3.9, 0.0, "heiter"),
],
}
# Für diese Städte klappt etwas nicht, so wie es bei echten Diensten vorkommt.
STOERUNGEN = {
"Bremen": 500, # der Dienst hat selbst einen Fehler
"Kiel": 503, # der Dienst wird gerade gewartet
"Rostock": "keine Antwort", # er nimmt die Anfrage an und antwortet nie
"Lübeck": "Abbruch", # er legt auf, ohne zu antworten
}
# Die Vorhersage gibt es nur mit Schlüssel in der Kopfzeile X-Api-Key. Wie jeder
# ordentliche Dienst kennt dieser den Schlüssel nicht selbst, sondern nur eine
# Prüfsumme davon.
PRUEFSUMME = "0fe6daa0a6754b2239b7022999ec7a5ae2b20fc096afc6d9eead51d07bef68ce"
class Anfrage(BaseHTTPRequestHandler):
def do_GET(self):
adresse = urlparse(self.path)
werte = {name: liste[0] for name, liste in parse_qs(adresse.query).items()}
stadt = werte.get("stadt", "")
stoerung = STOERUNGEN.get(stadt)
if adresse.path not in ("/wetter", "/vorhersage"):
self.sende(404, {"fehler": f"Unbekannte Adresse: {adresse.path}"})
elif stoerung == "keine Antwort":
# Wartet auf etwas, das nie passiert. Der Dienst endet trotzdem mit
# deinem Programm, weil er im Hintergrund läuft.
threading.Event().wait()
elif stoerung == "Abbruch":
self.close_connection = True
elif stoerung is not None:
self.sende(stoerung, self.responses[stoerung][0])
elif adresse.path == "/wetter":
self.wetter(stadt)
else:
self.vorhersage(stadt, werte.get("tage", "7"))
def wetter(self, stadt):
if stadt not in AKTUELL:
self.sende(404, {"fehler": f"Unbekannte Stadt: {stadt}"})
return
grad, wetter, wind = AKTUELL[stadt]
aktuell = {"temperatur": grad, "wetter": wetter, "wind": wind}
self.sende(200, {"stadt": stadt, "zeit": "2026-10-09T14:00", "aktuell": aktuell})
def vorhersage(self, stadt, tage):
schluessel = self.headers.get("X-Api-Key", "")
if hashlib.sha256(schluessel.encode()).hexdigest() != PRUEFSUMME:
self.sende(401, {"fehler": "Ohne gültigen Schlüssel in X-Api-Key gibt es keine Vorhersage"})
elif stadt not in VORHERSAGE:
self.sende(404, {"fehler": f"Keine Vorhersage für: {stadt}"})
elif not tage.isdigit() or not 1 <= int(tage) <= 7:
self.sende(400, {"fehler": "tage geht von 1 bis 7"})
else:
liste = []
for nummer, (hoch, tief, regen, wetter) in enumerate(VORHERSAGE[stadt][: int(tage)]):
liste.append(
{
"datum": (date(2026, 10, 9) + timedelta(days=nummer)).isoformat(),
"wetter": wetter,
"hoechstwert": hoch,
"tiefstwert": tief,
"regen": regen,
}
)
self.sende(200, {"stadt": stadt, "tage": liste})
def sende(self, status, inhalt):
"""Schickt die Antwort: ein Dictionary als JSON, einen Text als Text."""
if isinstance(inhalt, str):
art, rumpf = "text/plain", inhalt
else:
art, rumpf = "application/json", json.dumps(inhalt, ensure_ascii=False)
daten = rumpf.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", f"{art}; charset=utf-8")
self.send_header("Content-Length", str(len(daten)))
self.end_headers()
self.wfile.write(daten)
def log_message(self, *angaben):
# Sonst stünde jede Anfrage auf dem Fehlerkanal deines Programms.
pass
_server = None
def starten():
"""Startet den Dienst im Hintergrund. Er läuft, bis dein Programm endet."""
global _server
if _server is not None:
return
try:
_server = ThreadingHTTPServer(("127.0.0.1", 8000), Anfrage)
except OSError:
# Port 8000 ist belegt, weil der Dienst aus einem anderen Lauf noch
# antwortet. Dann fragt dein Programm eben den.
return
threading.Thread(target=_server.serve_forever, daemon=True).start() import os
import requests
import wetterdienst
wetterdienst.starten()
ADRESSE = "http://127.0.0.1:8000"
schluessel = os.environ["WETTER_SCHLUESSEL"]
print("Den Schlüssel gibt es, die Anfrage kann losgehen.") Das ist der erste Fehler, den du zu Hause zu sehen bekommst. Die Variable ist nicht gesetzt, und das
Programm endet mit KeyError: 'WETTER_SCHLUESSEL', wie bei jedem Dictionary, in dem ein Schlüssel
fehlt.
So soll es auch sein. Ein klarer Abbruch am Anfang ist besser als eine Anfrage ohne Schlüssel und
ein 401 später. Mit os.environ.get(...) gäbe es dagegen None, requests ließe die Kopfzeile
einfach weg, und der Fehler käme erst beim Dienst an.
Zu Hause setzt du die Variable im Terminal, bevor du das Programm startest. Unter Windows geht das
in der PowerShell mit $env:WETTER_SCHLUESSEL = "…", auf dem Mac und unter Linux mit
export WETTER_SCHLUESSEL=…. Das gilt für dieses eine Fenster, bis es geschlossen wird. Dauerhaft
geht es über die Systemeinstellungen von Windows oder über die Datei, die dein Terminal beim Start
liest, auf dem Mac meistens ~/.zshrc.
Abholen und aufheben
# Der Wetterdienst für diesen Abschnitt. Die Maschine hier hat kein Netz, deshalb
# läuft der Dienst auf derselben Maschine wie dein Programm, unter der Adresse
# http://127.0.0.1:8000. Die Anfragen an ihn sind trotzdem echt: Sie gehen über
# HTTP hinaus und kommen als Antwort zurück, nur eben nicht durch das Internet.
# Hier musst du nichts ändern.
import hashlib
import json
import threading
from datetime import date, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
# Das Wetter am 09.10.2026 um 14 Uhr: Grad, Wetter, Wind in km/h.
AKTUELL = {
"Berlin": (14.2, "bewölkt", 12),
"Hamburg": (12.8, "Regen", 24),
"München": (9.6, "Nebel", 3),
"Köln": (15.1, "heiter", 8),
"Frankfurt am Main": (13.4, "bewölkt", 10),
"Stuttgart": (11.9, "Regen", 14),
"Hannover": (13.1, "bewölkt", 16),
"Nürnberg": (10.8, "Nebel", 4),
"Leipzig": (13.7, "heiter", 9),
"Dresden": (12.3, "bewölkt", 11),
"Erfurt": (12.6, "heiter", 7),
}
# Die Vorhersage ab dem 09.10.2026, je Tag: Höchstwert, Tiefstwert, Regen in mm
# und das Wetter.
VORHERSAGE = {
"Leipzig": [
(15.2, 6.1, 0.0, "heiter"),
(17.8, 7.4, 0.0, "sonnig"),
(18.4, 8.0, 0.0, "sonnig"),
(16.9, 9.2, 0.4, "bewölkt"),
(13.5, 8.8, 6.2, "Regen"),
(11.2, 7.5, 9.8, "Regen"),
(12.6, 5.9, 1.1, "bewölkt"),
],
"Dresden": [
(14.6, 5.2, 0.2, "bewölkt"),
(16.3, 6.8, 0.0, "heiter"),
(15.0, 8.1, 0.0, "bewölkt"),
(16.3, 7.7, 0.0, "heiter"),
(19.1, 8.4, 0.0, "sonnig"),
(12.4, 7.0, 5.1, "Regen"),
(10.8, 6.2, 7.6, "Regen"),
],
"Erfurt": [
(13.9, 4.8, 0.0, "heiter"),
(15.6, 5.5, 0.0, "sonnig"),
(14.1, 7.2, 2.3, "Regen"),
(12.0, 6.6, 4.0, "Regen"),
(12.8, 5.1, 0.6, "bewölkt"),
(14.7, 4.3, 0.0, "heiter"),
(13.3, 3.9, 0.0, "heiter"),
],
}
# Für diese Städte klappt etwas nicht, so wie es bei echten Diensten vorkommt.
STOERUNGEN = {
"Bremen": 500, # der Dienst hat selbst einen Fehler
"Kiel": 503, # der Dienst wird gerade gewartet
"Rostock": "keine Antwort", # er nimmt die Anfrage an und antwortet nie
"Lübeck": "Abbruch", # er legt auf, ohne zu antworten
}
# Die Vorhersage gibt es nur mit Schlüssel in der Kopfzeile X-Api-Key. Wie jeder
# ordentliche Dienst kennt dieser den Schlüssel nicht selbst, sondern nur eine
# Prüfsumme davon.
PRUEFSUMME = "0fe6daa0a6754b2239b7022999ec7a5ae2b20fc096afc6d9eead51d07bef68ce"
class Anfrage(BaseHTTPRequestHandler):
def do_GET(self):
adresse = urlparse(self.path)
werte = {name: liste[0] for name, liste in parse_qs(adresse.query).items()}
stadt = werte.get("stadt", "")
stoerung = STOERUNGEN.get(stadt)
if adresse.path not in ("/wetter", "/vorhersage"):
self.sende(404, {"fehler": f"Unbekannte Adresse: {adresse.path}"})
elif stoerung == "keine Antwort":
# Wartet auf etwas, das nie passiert. Der Dienst endet trotzdem mit
# deinem Programm, weil er im Hintergrund läuft.
threading.Event().wait()
elif stoerung == "Abbruch":
self.close_connection = True
elif stoerung is not None:
self.sende(stoerung, self.responses[stoerung][0])
elif adresse.path == "/wetter":
self.wetter(stadt)
else:
self.vorhersage(stadt, werte.get("tage", "7"))
def wetter(self, stadt):
if stadt not in AKTUELL:
self.sende(404, {"fehler": f"Unbekannte Stadt: {stadt}"})
return
grad, wetter, wind = AKTUELL[stadt]
aktuell = {"temperatur": grad, "wetter": wetter, "wind": wind}
self.sende(200, {"stadt": stadt, "zeit": "2026-10-09T14:00", "aktuell": aktuell})
def vorhersage(self, stadt, tage):
schluessel = self.headers.get("X-Api-Key", "")
if hashlib.sha256(schluessel.encode()).hexdigest() != PRUEFSUMME:
self.sende(401, {"fehler": "Ohne gültigen Schlüssel in X-Api-Key gibt es keine Vorhersage"})
elif stadt not in VORHERSAGE:
self.sende(404, {"fehler": f"Keine Vorhersage für: {stadt}"})
elif not tage.isdigit() or not 1 <= int(tage) <= 7:
self.sende(400, {"fehler": "tage geht von 1 bis 7"})
else:
liste = []
for nummer, (hoch, tief, regen, wetter) in enumerate(VORHERSAGE[stadt][: int(tage)]):
liste.append(
{
"datum": (date(2026, 10, 9) + timedelta(days=nummer)).isoformat(),
"wetter": wetter,
"hoechstwert": hoch,
"tiefstwert": tief,
"regen": regen,
}
)
self.sende(200, {"stadt": stadt, "tage": liste})
def sende(self, status, inhalt):
"""Schickt die Antwort: ein Dictionary als JSON, einen Text als Text."""
if isinstance(inhalt, str):
art, rumpf = "text/plain", inhalt
else:
art, rumpf = "application/json", json.dumps(inhalt, ensure_ascii=False)
daten = rumpf.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", f"{art}; charset=utf-8")
self.send_header("Content-Length", str(len(daten)))
self.end_headers()
self.wfile.write(daten)
def log_message(self, *angaben):
# Sonst stünde jede Anfrage auf dem Fehlerkanal deines Programms.
pass
_server = None
def starten():
"""Startet den Dienst im Hintergrund. Er läuft, bis dein Programm endet."""
global _server
if _server is not None:
return
try:
_server = ThreadingHTTPServer(("127.0.0.1", 8000), Anfrage)
except OSError:
# Port 8000 ist belegt, weil der Dienst aus einem anderen Lauf noch
# antwortet. Dann fragt dein Programm eben den.
return
threading.Thread(target=_server.serve_forever, daemon=True).start() import json
import requests
import wetterdienst
wetterdienst.starten()
ADRESSE = "http://127.0.0.1:8000"
wetter = {}
for stadt in ["Berlin", "Hamburg", "Stuttgart", "Nürnberg"]:
antwort = requests.get(ADRESSE + "/wetter", params={"stadt": stadt}, timeout=2)
antwort.raise_for_status()
wetter[stadt] = antwort.json()["aktuell"]
with open("wetter.json", "w", encoding="utf-8") as datei:
json.dump(wetter, datei, indent=2, ensure_ascii=False)
# Ab hier braucht das Programm den Dienst nicht mehr.
with open("wetter.json", encoding="utf-8") as datei:
gespeichert = json.load(datei)
regen = [stadt for stadt, aktuell in gespeichert.items() if aktuell["wetter"] == "Regen"]
print(f"Regen in {len(regen)} von {len(gespeichert)} Städten:", ", ".join(regen)) Was einmal abgeholt ist, muss man nicht noch einmal holen. Hier wird das Wetter in vier Städten
abgefragt. Die Antworten landen in einem Dictionary, und das wird mit json.dump aus Lektion 12.5
als wetter.json abgelegt. Alles danach liest nur noch die Datei und sucht die Städte, in denen es
regnet. Zwei von vier.
Der zweite Teil liefe genauso, wenn der Dienst inzwischen nicht mehr erreichbar wäre. Was abgelegt
ist, zeigt cat wetter.json in der Befehlsleiste.
Bequemlichkeit ist nur ein Grund dafür. Viele Dienste begrenzen, wie oft man am Tag fragen darf, und wenn ein Programm bei jedem Probelauf neu fragt, ist man schnell darüber. Eine abgelegte Antwort lässt sich außerdem am nächsten Tag mit der neuen vergleichen. Und wer lieber in einer Tabelle nachsieht, schreibt die Daten als CSV wie in Lektion 12.3. Das passiert in der Aufgabe mit der Vorhersage.
Zu Hause
Auf deinem Rechner fallen wetterdienst.starten() und der Import weg, und in ADRESSE steht die
Adresse eines echten Dienstes. Den Wetterdienst hier gibt es nur in diesem Kurs. Echte Dienste haben
ihre eigenen Adressen, Parameter und Antworten, und die stehen in ihrer Beschreibung.
Ein Beispiel ist Bright Sky unter https://api.brightsky.dev. Dort gibt es die offenen Daten des
Deutschen Wetterdienstes als JSON, kostenlos und ohne Schlüssel. Gefragt wird nicht mit einem
Stadtnamen, sondern mit Koordinaten (lat und lon), und die Antwort sieht anders aus als hier.
Der Weg bleibt aber derselbe: Beschreibung lesen, eine Anfrage mit timeout, erst den Status, dann
die Daten, und zum Ansehen einmal json.dumps mit indent=2.
Bevor ein Programm jeden Tag von selbst fragt, lohnt noch ein Blick in die Nutzungsbedingungen. Da steht, wie oft man fragen darf und wofür man die Daten verwenden darf. Bei manchen Diensten steht dort auch, dass man sie nennen muss, wenn man die Daten weitergibt.
Was die Aufgabe will
Zum Schluss kommt der ganze Abschnitt in einem Programm zusammen. Es geht von den Einstellungen in einer Datei über den Schlüssel aus der Umgebung bis zur Tabelle am Ende, und die CSV, die dabei entsteht, kann danach auch eine Tabellenkalkulation lesen.
Zum Mitnehmen
Viele Dienste wollen einen Schlüssel (API key) sehen und antworten ohne ihn mit Status 401. Im Programm kommt der Schlüssel über os.environ aus einer Umgebungsvariablen und geht in einer Kopfzeile mit, nicht in der Adresse. Abgeholte Daten landen als JSON zum Weiterrechnen oder als CSV für die Tabellenkalkulation in einer Datei, und zu Hause steht statt des Wetterdienstes die Adresse eines echten Dienstes im Code.
Jetzt du
Basis Konto, kostenlosZu dieser Lektion gehört eine Aufgabe. Du schreibst den Code selbst, und nach jedem Lauf sagt dir eine Prüfliste, was schon stimmt.
Dafür brauchst du das Basis Konto. Es kostet nichts, und ein Passwort gibt es auch nicht.
In diesem Kurs läuft dein Code auf einem Server. Dafür hat das Basis Konto 1 Stunde im Monat, mehr Zeit gibt es mit dem Premium Konto.
Was in dieser Lektion steckt
-
Artikel mit 4 Beispielen zum Ausprobieren
Steht hier, ohne Konto lesbar.
-
Aufgabe, dein Code läuft auf einem Server
Öffnet sich mit dem Basis Konto.