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

Task_11: config.yaml

Browse Source
This commit is contained in:
Marco Schmid 2026-05-12 17:14:51 +02:00
parent bddf5a30cf
commit edf7893b27
5 changed files with 110 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
TASK.md
View File

@ -1,46 +1,66 @@
# Task 10 — Queries auslagern
# Task 11 — Externe Konfiguration mit config.yaml
## Rückblick Task 9: PoiType als StrEnum
## Rückblick Task 10: Queries auslagern
Ihr habt den Magic String `"bergbahn"` durch eine `StrEnum` ersetzt. Die wichtigsten Punkte:
Ihr habt die Overpass-Queries in eigene `.overpassql`-Dateien verschoben und
`load_query()` in `fetcher.py` eingeführt. Die wichtigsten Punkte:
- **Magic Strings** sind rohe Strings, die eine bestimmte Bedeutung haben, aber nirgends
formal definiert oder geprüft werden. Tippfehler wie `"Bergbahn"` oder `"berg_bahn"`
fallen erst zur Laufzeit auf — oder gar nicht.
- **`PoiType("bergbahn")`** funktioniert bei `StrEnum` — Python findet den passenden
Member anhand des Wertes. Das ist nützlich, wenn Werte aus einer Config oder
Datenbank kommen.
- **Grenzen:** Enums sind statisch — neue POI-Typen erfordern eine Code-Änderung.
Für sehr dynamische Systeme (Typen kommen aus der DB) wäre eine andere Lösung nötig.
- **Separation of Concerns:** Die Query ist jetzt klar vom Python-Code getrennt.
Jemand kann eine neue Query schreiben, ohne `fetcher.py` anzufassen — und umgekehrt
kann `fetcher.py` verbessert werden, ohne die Queries zu kennen.
- **`Path(__file__).parent / "queries"`** ist robuster als `"queries/"`, weil er
immer relativ zur Datei selbst aufgelöst wird — unabhängig davon, aus welchem
Verzeichnis das Skript gestartet wird.
- **Fehlender Query-File:** `load_query()` prüft explizit, ob die Datei existiert,
und wirft eine sprechende `OverpassApiError`. Ohne diese Prüfung käme ein
generischer `FileNotFoundError` — schwerer zu debuggen.
- **`{timeout}` und `{maxsize}` im Template:** Statt Werte im Query-String
hardzucoden, werden sie beim Laden eingefüllt. Das macht die Query flexibel
und die Werte zentral steuerbar.
## Aufgabe
In `main.py` steht aktuell die Overpass-Query als langer hardcodierter String direkt
im Code. Das hat mehrere Nachteile: Der String ist schwer lesbar, nicht wiederverwendbar,
und für jeden neuen POI-Typ muss `main.py` angefasst werden.
In `main.py` stehen noch immer Konfigurationswerte direkt im Code:
```python
BBOXEN = {
"davos": (46.72, 9.70, 46.92, 10.00),
"schweiz": (45.8, 5.9, 47.8, 10.5),
}
TIMEOUT = 25
MAXSIZE = 5000000
poi_type = PoiType.BERGBAHN
```
Wer eine neue Bbox hinzufügen oder einen anderen POI-Typ abfragen will,
muss Python-Code editieren. Das ist unpraktisch — und fehleranfällig.
**Konkret:**
1. Lege einen Unterordner `queries/` im Package an.
2. Verschiebe die Bergbahn-Query in eine Datei `queries/bergbahn.overpassql`.
Der Platzhalter `{bbox}` bleibt erhalten — ergänze zusätzlich `{timeout}` und
`{maxsize}` als Platzhalter (statt Hardcoding im Query-String).
3. Schreibe in `fetcher.py` eine neue Funktion `load_query()`:
```python
def load_query(poi_type: PoiType, bbox: tuple, timeout: int, maxsize: int) -> str:
...
```
Sie soll die passende `.overpassql`-Datei laden und alle Platzhalter befüllen.
4. Passe `_fetch_overpass()` an: Sie bekommt nun die **fertige Query** (kein
`bbox`-Parameter mehr) und sendet sie direkt.
5. Passe `load_pois()` und `main.py` entsprechend an.
1. Lege eine `config.yaml` im Package-Ordner an mit folgender Struktur:
```yaml
overpass:
timeout: 25
maxsize: 5000000
bboxen:
davos: [46.72, 9.70, 46.92, 10.00]
schweiz: [45.8, 5.9, 47.8, 10.5]
active_queries:
- bergbahn
```
2. Installiere `PyYAML` falls noch nicht vorhanden (`pip install pyyaml`).
3. Lese die Config in `main.py` mit `yaml.safe_load()` ein.
4. Ersetze alle hardcodierten Konstanten durch die Werte aus der Config.
5. Erzeuge `poi_types` als Liste von `PoiType`-Objekten aus `active_queries`
und iteriere in `main()` darüber. Damit können wir nachher nicht nur z.B. Restaurants fetchen, sondern zusätzlich
auch andere POI-Typen.
**Fragen zum Nachdenken:**
- Welche Vorteile hat es, Queries in eigenen Dateien zu speichern?
- Warum ist `Path(__file__).parent / "queries"` besser als ein relativer Pfad `"queries/"`?
- Was passiert, wenn die `.overpassql`-Datei für einen bestimmten `PoiType` fehlt —
wie sollte `load_query()` damit umgehen?
- Welches Design-Prinzip steckt hinter dieser Änderung?
(Stichwort: *Separation of Concerns*)
- Welche Arten von Konfiguration gehören in eine YAML-Datei, welche eher
in Umgebungsvariablen (`.env`)?
- Was passiert, wenn jemand in `active_queries` einen ungültigen Wert einträgt
(z.B. `"gondelbahn"`), der nicht im `PoiType`-Enum existiert?

144
src/overpass/fetcher.py
View File

@ -1,128 +1,52 @@
import logging
from pathlib import Path
import requests
from models import POI
logger = logging.getLogger(__name__)
from .models import PoiType, POI
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
# REMARK:
# zwei Strategien:
# Fail-fast: Ein Fehler bricht alles ab → sinnvoll, wenn jedes Element kritisch ist
# Best-effort: Fehlerhafte Elemente überspringen, Rest verarbeiten → sinnvoll bei OSM-Daten, wo einzelne Einträge unvollständig sein können
QUERY_DIR = Path(__file__).parent / "queries"
class OverpassApiError(Exception):
pass
def load_query(poi_type: PoiType, bbox: tuple, timeout: int, maxsize: int) -> str:
"""Lädt das Query-Template und füllt alle Platzhalter."""
query_file = QUERY_DIR / f"{poi_type}.overpassql"
if not query_file.exists():
raise OverpassApiError(f"Keine Query-Datei für '{poi_type}': {query_file}")
template = query_file.read_text(encoding="utf-8")
bbox_str = ",".join(map(str, bbox))
return template.format(bbox=bbox_str, timeout=timeout, maxsize=maxsize)
def load_pois(overpass_query: str, bbox: tuple, poi_type: str) -> list[POI]:
"""Führt Fetch und Parse zusammen aus."""
raw = _fetch_overpass(overpass_query=overpass_query, bbox=bbox)
return _parse_pois(raw, poi_type)
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)
def load_pois(query: str, poi_type: PoiType) -> list[POI]:
"""Führt die fertige Query aus und gibt POI-Objekte zurück."""
data = _fetch_overpass(query)
if "remark" in data:
raise OverpassApiError(f"Overpass Query-Fehler: {data['remark']}")
return _parse_pois(data, poi_type)
def _fetch_overpass(query: str) -> dict:
"""Schicht 1: nur HTTP – kein Parsing, kein POI-Wissen."""
try:
response = requests.post(
OVERPASS_URL,
data={"data": query},
timeout=15,
timeout=60,
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
response.raise_for_status()
except requests.RequestException as exc:
raise OverpassApiError("Overpass-API Request fehlgeschlagen") from exc
raise OverpassApiError(f"HTTP-Fehler: {exc}") from exc
return response.json()
data = response.json()
def _parse_pois(data: dict, poi_type: PoiType) -> list[POI]:
return [_parse_poi(el, poi_type) for el in data.get("elements", [])]
# zusätzliche Fehlermöglichkeit -> Status ist zwar 200, aber Liste mit Ergebnissen ist leer...
if "remark" in data:
raise OverpassApiError(f"Overpass Query-Fehler: {data['remark']}")
return data
def _parse_poi(data: dict, poi_type: str) -> POI:
""" Wandelt ein einzelnes Overpass-Element in ein POI-Objekt um.
:param data: dictionary mit Daten für ein POI-Objekt
:param poi_type: Bezeichnung für die geladene POI-Gruppe (z.B. 'restaurant', 'bergbahn', ...)
:return: POI-Objekt
"""
try:
return POI(
id=data['id'],
poi_type=poi_type,
type=data.get('type', ''),
lat=float(data.get("lat") or data["center"]["lat"]),
lon=float(data.get("lon") or data["center"]["lon"]),
tags=data.get('tags', {}),
)
except KeyError as exc:
raise OverpassApiError("Feld in API - Antwort fehlt") from exc
except (TypeError, ValueError) as exc:
raise OverpassApiError("API - Antwort hat falsches Format ") from exc
def _parse_pois(raw: dict, poi_type: str) -> list[POI]:
"""Extrahiert alle Elemente aus der API-Antwort und parst sie.
Fehlerhafte Elemente werden übersprungen und geloggt.
"""
pois = []
for element in raw.get("elements", []):
try:
pois.append(_parse_poi(data=element, poi_type=poi_type))
except OverpassApiError as exc:
logger.warning(f"POI übersprungen (id={element.get('id', '?')}): {exc}")
return pois
def _parse_poi(element: dict, poi_type: PoiType) -> POI:
return POI(
id = element["id"],
type = element.get("type", ""),
lat = element.get("lat") or element["center"]["lat"],
lon = element.get("lon") or element["center"]["lon"],
tags = element.get("tags", {}),
poi_type = str(poi_type),
)

22
src/overpass/main.py
View File

@ -1,5 +1,5 @@
import logging
from .fetcher import load_pois, OverpassApiError
from .fetcher import load_query, load_pois, OverpassApiError
from .models import POI, PoiType
logging.basicConfig(
@ -9,29 +9,19 @@ logging.basicConfig(
)
logger = logging.getLogger(__name__)
poi_type = PoiType.BERGBAHN
BBOXEN = {
"davos": (46.72, 9.70, 46.92, 10.00),
"schweiz": (45.8, 5.9, 47.8, 10.5),
}
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;
"""
TIMEOUT = 25
MAXSIZE = 5000000
poi_type = PoiType.BERGBAHN
def main() -> None:
for name, bbox in BBOXEN.items():
try:
pois: list[POI] = load_pois(overpass_query=QUERY, bbox=bbox, poi_type=poi_type)
query = load_query(poi_type, bbox, TIMEOUT, MAXSIZE)
pois: list[POI] = load_pois(query=query, poi_type=poi_type)
except OverpassApiError as exc:
logger.error(f"Fehler bei '{name}': {exc}")
continue

9
src/overpass/queries/bergbahn.overpassql Normal file
View File

@ -0,0 +1,9 @@
[out:json][timeout:{timeout}][maxsize:{maxsize}];
(
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;

8
src/overpass/queries/restaurant.overpassql Normal file
View File

@ -0,0 +1,8 @@
[out:json][timeout:{timeout}][maxsize:{maxsize}];
(
node["amenity"="restaurant"]({bbox});
way["amenity"="restaurant"]({bbox});
node["amenity"="cafe"]({bbox});
way["amenity"="cafe"]({bbox});
);
out center body;