schmidmarco/overpass
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Task_4: Adapter and Dataclass

Browse Source
This commit is contained in:
Marco Schmid 2026-04-23 18:11:43 +02:00
parent bb4a2b1131
commit bc9dde71ec
4 changed files with 129 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
TASK.md
View File

@ -1,7 +1,8 @@
# TASK 3:
# TASK 4:
* verlagert als Nächstes die Logik von 'fetch_bergbahnen' in ein eigenes Modul `overpass.py` aus.
* nennt die Funktion allgemeiner `fetch_overpass` (anstelle `fetch_bergbahnen`) -> somit würde es Sinn
machen, wenn wir den Query der Funktion als Argument mitgeben könnten (ist genereller).
* Erstellt in diesem `main.py` eine eigene 'main-Funktion', welche nur die Hauptlogik beinhalten und somit `fetch_overpass`
importiert und aufruft.
Wir arbeiten nun intern direkt mit den Daten von Overpass. Macht das Sinn? Warum vielleicht nicht?
* Versucht einen Adapter zu bauen, wir wollen intern mit einer eigenen Dataclass `POI` arbeiten. Wir bauen also dazu eine
Funktion `load_pois`, welche einerseits die Daten fetched und andererseits auch parsed. Den Fetching-Teil haben wir
bereits (`fetch_overpass`), den Pasing-Teil haben wir noch nicht.
Schreibt bitte eine eigene Dataclass`Poi` in welche die gefetchten Daten 'abgefüllt' werden können.

BIN
__pycache__/overpass.cpython-313.pyc Normal file
View File

Binary file not shown.

97
main.py
View File

@ -1,87 +1,40 @@
import requests
from pprint import pprint
from overpass import fetch_overpass, OverpassApiError
# ---------------------------------------------------------------------------
# Konfiguration
# ---------------------------------------------------------------------------
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
BBOXEN = {
"davos": (46.72, 9.70, 46.92, 10.00),
"schweiz": (45.8, 5.9, 47.8, 10.5),
}
BERGBAHN_QUERY = """
[out:json][timeout:3][maxsize:500000];
(
QUERY = """
[out:json][timeout:2][maxsize:500000];
(
node["aerialway"="station"]({bbox});
way["aerialway"="station"]({bbox});
node["railway"="funicular"]({bbox});
way["railway"="funicular"]({bbox});
node["railway"="station"]["funicular"="yes"]({bbox});
);
out center body;
"""
class OverpassApiError(Exception):
pass
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.
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:
OverpassApiError: Wenn die API nicht innerhalb des gesetzten Timeouts
antwortet (clientseitig, unabhängig vom serverseitigen
Timeout im Query).
OverpassApiError: Wenn der Request aus einem anderen Grund fehlschlägt
(z.B. 429 Too Many Requests, 504 Gateway Timeout,
Netzwerkfehler).
);
out center body;
"""
bbox_str = ",".join(map(str, bbox))
query = BERGBAHN_QUERY.format(bbox=bbox_str)
# ---------------------------------------------------------------------------
# Hauptlogik
# ---------------------------------------------------------------------------
def main() -> None:
for name, bbox in BBOXEN.items():
try:
response = requests.post(
OVERPASS_URL,
data={"data": query},
timeout=5,
headers={"User-Agent": "CDS Exercise"},
)
response.raise_for_status() # prüft den HTTP-Statuscode der Antwort und wirft eine Exception, wenn es ein Fehler war (requests.HTTPError)
except requests.Timeout as exc:
raise OverpassApiError("Overpass-API Timeout") from exc
except requests.RequestException as exc:
raise OverpassApiError("Overpass-API Request fehlgeschlagen") from exc
return response.json()
result = fetch_overpass(overpass_query=QUERY, bbox=bbox)
except OverpassApiError as exc:
print(f" Fehler : {exc}")
continue
elements = result.get("elements", [])
print(elements)
if __name__ == "__main__":
# bbox = (46.72, 9.70, 46.92, 10.00)
bbox = (45.8, 5.9, 47.8, 10.5)
result = fetch_bergbahnen(bbox)
pprint(result)
main()

90
overpass.py Normal file
View File

@ -0,0 +1,90 @@
import requests
from pprint import pprint
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
class OverpassApiError(Exception):
pass
def fetch_overpass(overpass_query: str, 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.
Args:
overpass_query (str): Overpass-QL-Query mit dem Platzhalter {bbox}.
Beispiel:
'[out:json][timeout:5];
(node["aerialway"="station"]({bbox}););
out center body;'
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:
OverpassApiError: Wenn die API nicht innerhalb des gesetzten Timeouts
antwortet (clientseitig, unabhängig vom serverseitigen
Timeout im Query).
OverpassApiError: Wenn der Request aus einem anderen Grund fehlschlägt
(z.B. 429 Too Many Requests, 504 Gateway Timeout,
Netzwerkfehler).
"""
bbox_str = ",".join(map(str, bbox))
query = overpass_query.format(bbox=bbox_str)
try:
response = requests.post(
OVERPASS_URL,
data={"data": query},
timeout=5,
headers={"User-Agent": "CDS Exercise"},
)
response.raise_for_status() # prüft den HTTP-Statuscode der Antwort und wirft eine Exception, wenn es ein Fehler war (requests.HTTPError)
except requests.Timeout as exc:
raise OverpassApiError("Overpass-API Timeout") from exc
except requests.RequestException as exc:
raise OverpassApiError("Overpass-API Request fehlgeschlagen") from exc
return response.json()
if __name__ == "__main__":
BERGBAHN_QUERY = """
[out:json][timeout:2][maxsize:500000];
(
node["aerialway"="station"]({bbox});
way["aerialway"="station"]({bbox});
node["railway"="funicular"]({bbox});
way["railway"="funicular"]({bbox});
node["railway"="station"]["funicular"="yes"]({bbox});
);
out center body;
"""
bbox = (46.72, 9.70, 46.92, 10.00)
# bbox = (45.8, 5.9, 47.8, 10.5)
result = fetch_overpass(overpass_query=BERGBAHN_QUERY, bbox=bbox)
pprint(result)