Caching & Monitoring
404-Fallbacks und Validierungsfehler bei Flughäfen: Fehlerbehandlung. Plattformweite HTTP-Fehler und Wiederholungsversuche: Fehlerbehandlung.
Datenaktualität
Flughafendatensätze, Daten zu Funknavigationsanlagen (Navaids) und Suchindizes werden periodisch aktualisiert — nicht in Echtzeit. Cachen Sie großzügig; Flughafen-Metadaten ändern sich von Tag zu Tag nur selten.
| Endpoint | Empfohlene TTL | Hinweise |
|---|---|---|
GET /airports/search | 24 Stunden | Statische Metadaten: Start- und Landebahnen, Frequenzen, Höhe |
GET /airports/search/text | 1 Stunde | Der Suchindex ist stabil; die Ergebnisse ändern sich innerhalb einer Sitzung nicht |
GET /navaids | 6 Stunden | Navaid-Daten werden im Rhythmus der Luftfahrt-Aktualisierungszyklen aktualisiert |
GET /airports/search/location | 15 Minuten | Ergebnisse zum nächstgelegenen Flughafen hängen vom Standort des Nutzers ab; bei geänderten Koordinaten erneut abfragen |
Caching
Verwenden Sie für Integrationen mit geringem Volumen einen einfachen prozessinternen TTL-Cache. Bei Deployments mit mehreren Instanzen ersetzen Sie das Dict durch Redis oder Memcached, damit der Cache gemeinsam genutzt wird und Neustarts übersteht.
import os
import time
import requests
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
_cache: dict[str, tuple[dict, float]] = {}
def cached_get(url: str, params: dict, ttl_seconds: int) -> dict | None:
"""Fetch with a simple in-process TTL cache. Returns None on 404."""
key = url + str(sorted(params.items()))
cached = _cache.get(key)
if cached and time.time() < cached[1]:
return cached[0]
r = requests.get(url, headers=HEADERS, params=params, timeout=(10, 15))
if r.status_code == 404:
return None
r.raise_for_status()
data = r.json()
_cache[key] = (data, time.time() + ttl_seconds)
return data
def get_airport(icao: str) -> dict | None:
return cached_get(f"{BASE}/airports/search", {"icao": icao}, ttl_seconds=86400)
def search_airports(query: str) -> list[dict]:
data = cached_get(f"{BASE}/airports/search/text", {"q": query}, ttl_seconds=3600)
return (data or {}).get("airports") or []
def get_navaids(icao: str) -> list[dict]:
data = cached_get(f"{BASE}/navaids", {"airport": icao}, ttl_seconds=21600)
return (data or {}).get("navaids") or []Rate Limits
Jede Antwort enthält Rate-Limit-Header. Kontingentstufen und das Muster für Wiederholungsversuche bei 429 finden Sie unter Fehlerbehandlung.
| Header | Wert |
|---|---|
X-RateLimit-Requests-Limit | Monatliches Anfragekontingent Ihres Tarifs |
X-RateLimit-Requests-Remaining | Verbleibende Anfragen in diesem Monat |
X-RateLimit-Requests-Reset | Sekunden bis zum Zurücksetzen des Kontingents |
Da Flughafendaten eine lange TTL haben, verbraucht eine gut gecachte Integration pro Sitzung nur sehr wenige Anfragen. Eine einzelne /airports/search-Antwort für einen bestimmten ICAO-Code kann bedenkenlos für den Rest des Tages wiederverwendet werden.
Lauffähiges Beispiel
Ruft KJFK mit einem 24-Stunden-TTL-Cache ab und gibt bei jedem Aufruf Cache-Hit/-Miss aus.
import os
import time
import requests
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
_cache: dict[str, tuple[dict, float]] = {}
def get_airport(icao: str) -> dict | None:
url = f"{BASE}/airports/search"
key = f"airport:{icao}"
cached = _cache.get(key)
if cached and time.time() < cached[1]:
print(f" cache HIT — {icao}")
return cached[0]
print(f" cache MISS — {icao}, fetching...")
r = requests.get(url, headers=HEADERS, params={"icao": icao}, timeout=(10, 15))
if r.status_code == 404:
return None
r.raise_for_status()
data = r.json()
_cache[key] = (data, time.time() + 86400)
return data
if __name__ == "__main__":
airport = get_airport("KJFK")
if airport:
print(f" {airport.get('name')} — elevation {airport.get('elevation_ft')} ft")
# Second call — served from cache
airport = get_airport("KJFK")
if airport:
print(f" {airport.get('name')} — served from cache")Monitoring
Protokollieren Sie bei jeder Anfrage an die Flughafen-API diese Felder:
import logging
import time
import requests
logger = logging.getLogger("airport")
HEADERS = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
"X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"
def fetch_airport_logged(icao: str) -> dict | None:
start = time.monotonic()
r = requests.get(
f"{BASE}/airports/search",
headers=HEADERS,
params={"icao": icao},
timeout=(10, 15),
)
elapsed_ms = (time.monotonic() - start) * 1000
log_data = {
"endpoint": "airports/search",
"icao": icao,
"status": r.status_code,
"elapsed_ms": round(elapsed_ms),
"quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"),
}
if r.status_code == 200:
logger.info("airport_fetch", extra=log_data)
return r.json()
elif r.status_code == 404:
logger.info("airport_not_found", extra=log_data)
return None
else:
logger.warning("airport_error", extra=log_data)
r.raise_for_status()Alarmieren bei:
status = 429— Ihre Abfragerate übersteigt Ihren Tarif; TTL erhöhen oder Tarif upgradenelapsed_ms > 5000wiederholt — Latenz beim vorgelagerten Dienst; Timeout und Fallback in Betracht ziehenquota_remaining < 200— Sie nähern sich Ihrem monatlichen Limit
Siehe auch: Flughäfen · Flughafensuche · Navaids · Fehlerbehandlung