Task 3: overpass modul

This commit is contained in:
Marco Schmid 2026-04-21 18:43:22 +02:00
parent e2bbbf8e2f
commit df15c3c988

80
main.py
View File

@ -17,30 +17,92 @@ out center body;
""" """
def fetch_bergbahnen(bbox) -> dict: def fetch_bergbahnen(bbox: tuple) -> dict:
"""
Fragt die Overpass API nach Bergbahnen in der angegebenen Bounding Box ab.
Sendet einen HTTP-POST-Request an die Overpass API und gibt die geparste
JSON-Antwort zurück. Die Funktion prüft den HTTP-Status-Code, bevor sie
versucht die Antwort als JSON zu parsen.
Args:
bbox (tuple): Bounding Box als 4-Tuple in Dezimalgrad:
(south, west, north, east)
Beispiel Davos: (46.72, 9.70, 46.92, 10.00)
Beispiel Schweiz: (45.8, 5.9, 47.8, 10.5)
Returns:
dict: Geparste JSON-Antwort der Overpass API. Die Antwort enthält
unter dem Schlüssel "elements" eine Liste von OSM-Objekten
(nodes und ways) mit ihren Tags und Koordinaten.
Beispiel:
{
"elements": [
{
"type": "node",
"id": 123456,
"lat": 46.8, "lon": 9.8,
"tags": {"aerialway": "station", "name": "Jakobshorn"}
},
...
]
}
Raises:
RuntimeError: Wenn die API einen HTTP-Fehlercode zurückgibt
(z.B. 429 Too Many Requests, 504 Gateway Timeout).
RuntimeError: Wenn der Response-Body kein gültiges JSON enthält
(z.B. bei leerem Body nach einem Server-Timeout).
requests.Timeout: Wenn die API nicht innerhalb einer bestimmten Zeit
(timeout) antwortet.
"""
bbox_str = ",".join(map(str, bbox)) bbox_str = ",".join(map(str, bbox))
query = BERGBAHN_QUERY.format(bbox=bbox_str) query = BERGBAHN_QUERY.format(bbox=bbox_str)
resp = requests.post( resp = requests.post(
OVERPASS_URL, OVERPASS_URL,
data={"data": query}, data={"data": query},
timeout=90, timeout=10, # clientseitiges timeout ist nicht serverseitiges timeout (das ist im query selbst auf z.B. 3 Sekunden gestellt)
headers={"User-Agent": "GeoService/1.0 (poi-generator)"}, headers={"User-Agent": "GeoService/1.0 (poi-generator)"},
) )
return resp.json()
if resp.status_code != 200:
raise RuntimeError(
f"Overpass API Fehler: {resp.status_code}\n{resp.text}"
)
try:
data = resp.json()
except requests.exceptions.JSONDecodeError:
raise RuntimeError(
f"Antwort ist kein gültiges JSON:\n{resp.text[:200]}"
)
# HTTP 200 bedeutet nicht immer Erfolg — Overpass meldet Fehler im Body unter 'remark'!
if "remark" in data:
raise RuntimeError(
f"Overpass Query-Fehler: {data['remark']}"
)
return data
if __name__ == "__main__": if __name__ == "__main__":
bbox = (46.72, 9.70, 46.92, 10.00) bbox = (46.72, 9.70, 46.92, 10.00)
# bbox = (45.8, 5.9, 47.8, 10.5)
result = fetch_bergbahnen(bbox) result = fetch_bergbahnen(bbox)
pprint(result) pprint(result)
# TASK: # Was ist passiert?
# Wir haben verschiedene Fehler (verschiedene Ebenen) abgefangen:
# Ebene 1 — Netzwerk: requests.Timeout → Client wartet vergebens
# Ebene 2 — HTTP: status_code != 200 → Server meldet Fehler
# Ebene 3 — Fachlich: "remark" im JSON-Body → Query lief, aber mit Fehler
# * Fügt der aktuellen Funktion einen aussagekräftigen docstrings und typehints hinzu # TASK:
# * Versucht ein Error-Management mit aufzubauen. Werft einen 'RuntimeError'-Error, wenn etwas in der Funktion # * verlagert als nächstes die Logik von 'fetch_bergbahnen' in ein eigenes Modul 'overpass.py' aus.
# nicht klappt. # * nennt die Funktion allgemeiner 'fetch_overpass' (anstelle fetch_bergbahnen) -> somit würde es Sinn
# -> Was könnte nicht klappen? # machen, wenn wir den Query der Funktion als Argument mitgeben könnten (ist genereller).
# * Versucht mal bbox auf (45.8, 5.9, 47.8, 10.5) zu setzen und die Funktion aufzurufen # * Erstellt in diesem main.py eine eigene 'main-Funktion', welche nur die Hauptlogik beinhalten und somit 'fetch_overpass'
# importiert und aufruft.