# Caching & Monitoring

Muster für den Produktivbetrieb im Flugbetrieb – Caching, Rate Limits und Monitoring.



Fehlerbehandlung und Besonderheiten der Felder: [Fehlerbehandlung](/docs/v31/flight-operations-integration-errors). HTTP-Fehler der Plattform und Wiederholungsversuche: [Fehlerbehandlung](/docs/errors).

### Datenaktualität [#data-freshness]

Daten zum Flugbetrieb reichen von nahezu Echtzeit (Flugstatus) bis statisch (Entfernung). Passen Sie Ihre TTL an die Aktualisierungsfrequenz an — zu langes Caching von Live-Daten verschlechtert die UX, zu kurzes Caching statischer Daten verschwendet Kontingent.

| Endpoint                        | Empfohlene TTL                     | Hinweise                                                                                         |
| ------------------------------- | ---------------------------------- | ------------------------------------------------------------------------------------------------ |
| `GET /flight_status/{callsign}` | 30–60 Sekunden                     | Live-Tracking; Position und Status ändern sich jede Minute                                       |
| `GET /schedules`                | 2–5 Minuten                        | Flugpläne sind innerhalb einer Sitzung stabil, ändern sich aber bei Betriebsstörungen            |
| `GET /distance`                 | 24 Stunden                         | Die Großkreisentfernung ist für eine feste Route statisch                                        |
| `GET /briefing`                 | Nicht cachen (oder max. 5 Minuten) | Wird pro Anfrage von einer KI generiert; erneut abrufen, wenn der Benutzer explizit aktualisiert |

### Briefing-Latenz [#briefing-latency]

Der Briefing-Endpoint ruft ein KI-Inferenz-Backend ([IBM Granite](https://www.ibm.com/granite)) auf, um eine Zusammenfassung vor dem Abflug zu generieren. Rechnen Sie mit **60–90 Sekunden** Antwortzeit. Planen Sie dafür:

* Zeigen Sie sofort eine Ladeanzeige an, wenn der Benutzer ein Briefing anfordert.
* Verwenden Sie `timeout=(10, 90)` — das Verbindungs-Timeout kann bei 10 s bleiben, aber das Lese-Timeout muss verlängert werden.
* Wiederholen Sie die Anfrage bei `502`/`503` einmal, bevor Sie einen Fehler anzeigen.

### 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]] = {}


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_flight_status(callsign: str) -> dict | None:
    return cached_get(f"{BASE}/flight_status/{callsign}", {}, ttl_seconds=45)


def get_schedules(icao: str, direction: str = "departures") -> list[dict]:
    data = cached_get(
        f"{BASE}/schedules",
        {"airport": icao, "direction": direction},
        ttl_seconds=180,
    )
    return (data or {}).get("flights") or []


def get_distance(origin: str, destination: str) -> dict | None:
    return cached_get(
        f"{BASE}/distance",
        {"from_icao": origin, "to_icao": destination},
        ttl_seconds=86400,
    )
```

Ersetzen Sie `_cache` im Produktivbetrieb durch Redis oder Memcached, damit der Cache instanzübergreifend geteilt wird.

### Rate Limits [#rate-limits]

Jede Antwort enthält Rate-Limit-Header. Siehe [Fehlerbehandlung](/docs/errors) für Kontingentstufen und das Wiederholungsmuster bei 429.

| Header                           | Wert                                          |
| -------------------------------- | --------------------------------------------- |
| `X-RateLimit-Requests-Limit`     | Das monatliche Anfragekontingent Ihres Tarifs |
| `X-RateLimit-Requests-Remaining` | Verbleibende Anfragen in diesem Monat         |
| `X-RateLimit-Requests-Reset`     | Sekunden bis zum Zurücksetzen des Kontingents |

Polling des Flugstatus mit 45 Sekunden TTL kostet etwa 2 Anfragen/Minute pro verfolgtem Flug. Skalieren Sie Ihren Tarif entsprechend — für eine Abflugtafel mit 10 Flügen sind das \~2.880 Anfragen/Tag.

### Lauffähiges Beispiel [#runnable-example]

Ruft die Entfernung KJFK–KLAX mit einem TTL-Cache von 24 Stunden ab und gibt bei jedem Aufruf Treffer/Fehlschlag aus.

```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]] = {}


def get_distance(origin: str, destination: str) -> dict | None:
    key = f"distance:{origin}:{destination}"
    cached = _cache.get(key)
    if cached and time.time() < cached[1]:
        print(f"  cache HIT  — {origin}→{destination}")
        return cached[0]

    print(f"  cache MISS — {origin}→{destination}, fetching...")
    r = requests.get(
        f"{BASE}/distance",
        headers=HEADERS,
        params={"from_icao": origin, "to_icao": destination},
        timeout=(10, 15),
    )
    r.raise_for_status()
    data = r.json()
    _cache[key] = (data, time.time() + 86400)
    return data


if __name__ == "__main__":
    result = get_distance("KJFK", "KLAX")
    if result:
        nm = result.get("distance_nm") or result.get("distance")
        print(f"  KJFK → KLAX: {nm} nm")

    # Second call — served from cache
    result = get_distance("KJFK", "KLAX")
    if result:
        print(f"  served from cache")
```

### Monitoring [#monitoring]

Protokollieren Sie diese Felder bei jeder Anfrage zum Flugbetrieb:

```python
import logging
import time
import requests

logger = logging.getLogger("flight_ops")
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_flight_status_logged(callsign: str) -> dict | None:
    start = time.monotonic()
    r = requests.get(
        f"{BASE}/flight_status/{callsign}",
        headers=HEADERS,
        timeout=(10, 15),
    )
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "flight_status",
        "callsign": callsign,
        "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("flight_status_fetch", extra=log_data)
        return r.json()
    elif r.status_code == 404:
        logger.info("flight_not_found", extra=log_data)
        return None
    else:
        logger.warning("flight_status_error", extra=log_data)
        r.raise_for_status()
```

**Alarmieren bei:**

* `elapsed_ms > 5000` beim Flugstatus — Upstream-Latenz; Polling reduzieren oder einen Fallback hinzufügen
* `elapsed_ms > 95000` beim Briefing — Zeitüberschreitung bei der Generierung; dem Benutzer eine Wiederholungsoption anbieten
* `status = 429` — Polling-Rate übersteigt das Tarifkontingent; TTL erhöhen oder Tarif upgraden
* `quota_remaining < 500` — monatliches Limit bald erreicht; Polling-Frequenz überprüfen

***

Siehe auch: [Flugstatus](/docs/v31/flight-status-integration) · [Flugpläne](/docs/v31/schedules-integration) · [Entfernung](/docs/v31/distance-integration) · [Flugbriefing](/docs/v31/flight-briefing-integration) · [Fehlerbehandlung](/docs/v31/flight-operations-integration-errors)