# Caching & Monitoring Muster für den Produktivbetrieb mit NOTAMs und FAA-Verspätungen — Abfragerhythmus, Caching und Monitoring. Bedeutung leerer Listen und von null Warnungen: [Fehlerbehandlung](/docs/v31/alerts-integration-errors). Plattformweite HTTP-Fehler und Wiederholungsversuche: [Fehlerbehandlung](/docs/errors). ### Datenaktualität [#data-freshness] | Endpoint | Aktualisierungsrhythmus | Empfohlene TTL | Hinweise | | ------------------------ | ----------------------- | --------------- | ----------------------------------------------------------------------------- | | `GET /notams/{icao}` | Fortlaufend | **max. 5 Min.** | NOTAMs werden jederzeit herausgegeben, geändert und aufgehoben | | `GET /delays/faa` | Alle 2–5 Min. (ATCSCC) | 2–5 Min. | Landesweite Momentaufnahme; Ground Stops können ohne Vorankündigung auftreten | | `GET /delays/faa/{icao}` | Alle 2–5 Min. (ATCSCC) | 2–5 Min. | Flughafenspezifischer Ausschnitt desselben Feeds | **Cachen Sie bei keinem der beiden Endpoints länger als die empfohlene TTL.** Wird ein Ground Stop verhängt, während ein veralteter Cache-Eintrag ausgeliefert wird, sehen Ihre Nutzer während eines aktiven Programms „keine aktiven Verspätungen“. ### Caching [#caching] ```python 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]] = {} NOTAM_TTL = 300 # 5 minutes FAA_DELAY_TTL = 120 # 2 minutes def cached_get(url: str, ttl_seconds: int) -> dict | None: entry = _cache.get(url) if entry and time.time() < entry[1]: return entry[0] r = requests.get(url, headers=HEADERS, timeout=(10, 15)) if r.status_code == 404: return None r.raise_for_status() data = r.json() _cache[url] = (data, time.time() + ttl_seconds) return data def get_notams(icao: str) -> dict | None: """Cached NOTAM fetch — 5-minute TTL.""" return cached_get(f"{BASE}/notams/{icao}", NOTAM_TTL) def get_faa_delays_nationwide() -> dict | None: """Cached nationwide FAA delay snapshot — 2-minute TTL.""" return cached_get(f"{BASE}/delays/faa", FAA_DELAY_TTL) # data = get_notams("KJFK") # delays = get_faa_delays_nationwide() ``` Ersetzen Sie im Produktivbetrieb den Dict-Cache durch einen gemeinsamen Speicher (Redis, Memcached), damit der Cache über alle Instanzen hinweg konsistent ist und Neustarts übersteht. ### Rate Limits [#rate-limits] Jede Antwort enthält Rate-Limit-Header: | 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 | Tarifkontingente: | Tarif | Anfragen/Monat | | ----- | -------------- | | Free | 1.000 | | Basic | 5.000 | | Pro | 50.000 | | Ultra | 200.000 | | Mega | 600.000 | ### Monitoring und Alarmierung [#monitoring-and-alerting] Protokollieren Sie jede Anfrage zu [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) und FAA-Verspätungen und richten Sie aktive Alarme für betrieblich relevante Änderungen ein: ```python import logging import time import requests logger = logging.getLogger("alerts") 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" # Track previous counts to detect sudden changes _prev_notam_counts: dict[str, int] = {} _prev_faa_ground_stops: set[str] = set() WATCHED_AIRPORTS = ["KJFK", "KLAX", "KORD"] def check_notams_for_airport(icao: str) -> None: start = time.monotonic() r = requests.get(f"{BASE}/notams/{icao}", headers=HEADERS, timeout=(10, 15)) elapsed_ms = (time.monotonic() - start) * 1000 log_data = { "endpoint": "notams", "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: data = r.json() count = data.get("total", len(data.get("notams", []))) log_data["notam_count"] = count logger.info("notam_fetch", extra=log_data) prev = _prev_notam_counts.get(icao) if prev is not None and abs(count - prev) >= 5: logger.warning( "notam_count_jump", extra={"icao": icao, "prev": prev, "current": count}, ) _prev_notam_counts[icao] = count elif r.status_code == 404: logger.info("notam_not_found", extra=log_data) else: logger.warning("notam_error", extra=log_data) r.raise_for_status() def check_faa_ground_stops() -> None: start = time.monotonic() r = requests.get(f"{BASE}/delays/faa", headers=HEADERS, timeout=(10, 15)) elapsed_ms = (time.monotonic() - start) * 1000 log_data = { "endpoint": "delays/faa", "status": r.status_code, "elapsed_ms": round(elapsed_ms), "quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"), } if r.status_code == 200: data = r.json() total = data.get("total_alerts", 0) log_data["total_alerts"] = total logger.info("faa_delay_fetch", extra=log_data) # Alert when a ground stop appears at a watched airport current_gs = { gs["airport"] for gs in data.get("ground_stops", []) if gs.get("airport") in WATCHED_AIRPORTS } new_gs = current_gs - _prev_faa_ground_stops if new_gs: logger.warning( "ground_stop_appeared", extra={"airports": sorted(new_gs)}, ) _prev_faa_ground_stops.clear() _prev_faa_ground_stops.update(current_gs) else: logger.warning("faa_delay_error", extra=log_data) r.raise_for_status() # for icao in WATCHED_AIRPORTS: # check_notams_for_airport(icao) # check_faa_ground_stops() ``` **Alarmieren bei:** * `notam_count_jump` — die Anzahl der [NOTAMs](https://www.icao.int/safety/pages/notam-service-profiles.aspx) an einem überwachten Flughafen ändert sich zwischen zwei Abfragen um 5 oder mehr; ein ungewöhnlicher Anstieg kann auf Luftraumbeschränkungen oder Notfallmeldungen hindeuten * `ground_stop_appeared` — an einem Flughafen auf Ihrer Beobachtungsliste tritt ein Ground Stop in Kraft; Ground Stops können mit wenigen Minuten Vorlauf sämtliche Abflüge blockieren * `elapsed_ms > 5000` wiederholt — Latenzproblem beim vorgelagerten Dienst; Timeout und Fallback-UI in Betracht ziehen * `quota_remaining < 200` — nahe am monatlichen Limit; Abfragehäufigkeit reduzieren oder Tarif upgraden * `status == 429` mehr als zweimal innerhalb von 5 Minuten — Ihre Abfragerate ist zu hoch ### Lauffähiges Beispiel [#runnable-example] ```python 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[list, float]] = {} def get_notams(icao: str) -> list[dict]: key = icao.upper() hit = _cache.get(key) if hit and time.time() < hit[1]: return hit[0] r = requests.get(f"{BASE}/notams/{key}", headers=HEADERS, timeout=(10, 15)) r.raise_for_status() rows = r.json().get("notams") or [] _cache[key] = (rows, time.time() + 300) return rows if __name__ == "__main__": rows = get_notams("KJFK") print(f"KJFK NOTAMs: {len(rows)} (first call)") print(f"Cached repeat: {len(get_notams('KJFK'))} NOTAMs") ``` *** Siehe auch: [NOTAMs](/docs/v31/notams-integration) · [FAA-Verspätungen](/docs/v31/faa-delays-integration) · [Fehlerbehandlung](/docs/v31/alerts-integration-errors)