# Caching & Monitoring

Muster für den Produktivbetrieb mit ADS-B- und Luftfahrzeugdaten — Abfrageintervalle, Caching und Verwaltung des Kontingents.



404/422-Fallbacks und Umgang mit Tarifbeschränkungen: [Fehlerbehandlung](/docs/v31/aircraft-adsb-integration-errors). HTTP-Fehler der Plattform und Wiederholungsversuche: [Fehlerbehandlung](/docs/errors).

### Aktualität der Daten [#data-freshness]

| Endpoint                                    | Aktualisierungsrhythmus           | Empfohlene TTL     | Hinweise                                                                               |
| ------------------------------------------- | --------------------------------- | ------------------ | -------------------------------------------------------------------------------------- |
| `GET /aircraft/registration/{reg}`          | Register wird selten aktualisiert | 24 h               | Kennzeichen und Betreiber ändern sich selten                                           |
| `GET /aircraft/performance/{type}`          | Statische Spezifikationen         | 7 d                | Leistungswerte sind pro Typcode versionsgebunden                                       |
| `GET /adsb/aircraft`                        | Alle paar Sekunden                | **Niemals cachen** | Positionsdaten sind beim nächsten Rendering veraltet                                   |
| `GET /adsb/aircraft/statistics`             | Alle paar Sekunden                | **Niemals cachen** | Zählwerte über den gesamten Feed ändern sich laufend                                   |
| `GET /ultra/history/flights`                | Nur anhängendes Archiv            | 5–15 min           | Neue Flüge kommen nach der Landung hinzu, nicht in Echtzeit                            |
| `GET /ultra/history/airport/{icao}/traffic` | Nur anhängendes Archiv            | 5–15 min           | Dasselbe Archiv; Ergebnisse von Zeitraumabfragen sind innerhalb weniger Minuten stabil |

### Tarifstufen und Pfadpräfixe [#plan-tiers-and-path-prefixes]

Historische Endpoints sind an den Tarif gebunden. Das Pfadpräfix bestimmt den maximal verfügbaren Aufbewahrungszeitraum:

| Pfadpräfix           | Max. Zeitraum | Mindesttarif |
| -------------------- | ------------- | ------------ |
| `/ultra/history/...` | 90 Tage       | Pro          |
| `/mega/history/...`  | 365 Tage      | Ultra        |

Ein `401` oder `403` bei einem der beiden Präfixe bedeutet, dass der Tarif des Schlüssels unter der Schwelle liegt. Siehe [Fehlerbehandlung](/docs/v31/aircraft-adsb-integration-errors#historical-ads-b-plan-gating-and-window-errors) für das Muster mit dem message-Schlüssel.

### Caching [#caching]

Cachen Sie Abfragen zu Luftfahrzeugregister und Flugleistungsdaten aggressiv. Live-[ADS-B](https://www.faa.gov/air_traffic/technology/adsb)-Daten dürfen niemals gecacht werden — eine Position aus dem Cache auszuliefern, widerspricht dem Zweck der Echtzeitverfolgung.

```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:
    key = url + str(sorted(params.items()))
    entry = _cache.get(key)
    if entry and time.time() < entry[1]:
        return entry[0]

    r = requests.get(url, headers=HEADERS, params=params, timeout=(10, 15))
    if r.status_code in (404, 422):
        return None
    r.raise_for_status()
    data = r.json()
    _cache[key] = (data, time.time() + ttl_seconds)
    return data


def get_aircraft(registration: str) -> dict | None:
    """Cache registration lookups for 24 hours."""
    return cached_get(
        f"{BASE}/aircraft/registration/{registration}",
        {},
        ttl_seconds=86_400,
    )


def get_performance(icao_type: str) -> dict | None:
    """Cache performance specs for 7 days."""
    return cached_get(
        f"{BASE}/aircraft/performance/{icao_type}",
        {},
        ttl_seconds=604_800,
    )


def fetch_live_adsb(lat: float, lon: float, radius: int) -> list[dict]:
    # Do NOT cache — always fetch fresh positions.
    r = requests.get(
        f"{BASE}/adsb/aircraft",
        headers=HEADERS,
        params={"lat": lat, "lon": lon, "radius": radius, "limit": 50},
        timeout=(10, 15),
    )
    r.raise_for_status()
    return r.json().get("aircraft", [])


# ac = get_aircraft("N636JB")
# if ac and ac.get("found"):
#     perf = get_performance(ac["aircraft"]["icao_type"])
#
# aircraft = fetch_live_adsb(40.6413, -73.7781, 80)  # always live
```

Ersetzen Sie für den Produktivbetrieb den Dict-Cache durch einen gemeinsam genutzten Speicher (Redis, Memcached), damit der Cache Neustarts übersteht und von allen Instanzen gemeinsam genutzt wird.

### Hinweise zum ADS-B-Polling [#ads-b-polling-guidance]

Live-[ADS-B](https://www.faa.gov/air_traffic/technology/adsb)-Daten werden auf Feed-Seite alle paar Sekunden aktualisiert. Sinnvolle Abfrageintervalle auf Client-Seite:

| Anwendungsfall               | Empfohlenes Abfrageintervall |
| ---------------------------- | ---------------------------- |
| Bewegte Kartenanzeige        | 5–15 Sekunden                |
| Widget zur Verkehrsübersicht | 15–30 Sekunden               |
| Zähler-Badge im Dashboard    | 60 Sekunden                  |

**Fragen Sie nicht bei jedem Rendering ab.** Wenn Ihre Komponente bei Mausbewegungen oder Scroll-Ereignissen neu gerendert wird, entkoppeln Sie den [ADS-B](https://www.faa.gov/air_traffic/technology/adsb)-Abruf vollständig vom Rendering-Zyklus — verwenden Sie ein Hintergrundintervall oder einen Datenspeicher.

### 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 |

Kontingente der Tarife:

| Tarif | Anfragen/Monat |
| ----- | -------------- |
| Free  | 1.000          |
| Basic | 5.000          |
| Pro   | 50.000         |
| Ultra | 200.000        |
| Mega  | 600.000        |

### Monitoring [#monitoring]

Protokollieren Sie diese Werte bei jeder [ADS-B](https://www.faa.gov/air_traffic/technology/adsb)- und Luftfahrzeuganfrage:

```python
import logging
import time
import requests

logger = logging.getLogger("aircraft")

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_live_adsb_logged(lat: float, lon: float, radius: int) -> list[dict]:
    start = time.monotonic()
    r = requests.get(
        f"{BASE}/adsb/aircraft",
        headers=HEADERS,
        params={"lat": lat, "lon": lon, "radius": radius, "limit": 100},
        timeout=(10, 15),
    )
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "adsb/aircraft",
        "lat": lat,
        "lon": lon,
        "radius": radius,
        "status": r.status_code,
        "elapsed_ms": round(elapsed_ms),
        "quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"),
    }

    if r.status_code == 200:
        aircraft = r.json().get("aircraft", [])
        log_data["aircraft_count"] = len(aircraft)
        logger.info("adsb_fetch", extra=log_data)
        return aircraft

    logger.warning("adsb_error", extra=log_data)
    r.raise_for_status()
    return []
```

**Alarmieren bei:**

* `aircraft_count == 0` bei mehr als drei aufeinanderfolgenden Abfragen für dasselbe Gebiet — kann auf einen Ausfall des Feeds statt auf tatsächlich leeren Luftraum hindeuten
* `elapsed_ms > 5000` wiederholt — Latenzproblem auf Upstream-Seite
* `quota_remaining < 200` — das Monatslimit ist bald erreicht; reduzieren Sie die Abfragefrequenz oder wechseln Sie in einen höheren Tarif
* `status == 429` mehr als zweimal innerhalb von 5 Minuten — Ihr Abfrageintervall ist für Ihren Tarif zu aggressiv

### Ausführbares 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[dict, float]] = {}


def get_aircraft_cached(registration: str) -> dict | None:
    key = registration.upper()
    hit = _cache.get(key)
    if hit and time.time() < hit[1]:
        return hit[0]
    r = requests.get(
        f"{BASE}/aircraft/registration/{key}",
        headers=HEADERS,
        timeout=(10, 15),
    )
    if r.status_code == 404:
        return None
    r.raise_for_status()
    data = r.json()
    _cache[key] = (data, time.time() + 86_400)
    return data


if __name__ == "__main__":
    ac = get_aircraft_cached("N636JB")
    found = ac and ac.get("found")
    print(f"N636JB lookup: {'found' if found else 'not found'}")
    ac2 = get_aircraft_cached("N636JB")
    print("Second call served from 24 h cache:", ac2 is ac)
```

***

Siehe auch: [Luftfahrzeugabfrage](/docs/v31/aircraft-lookup-integration) · [Live-Tracking](/docs/v31/adsb-integration) · [Fehlerbehandlung](/docs/v31/aircraft-adsb-integration-errors)