Caching & Monitoring

404/422-Fallbacks und Umgang mit Tarifbeschränkungen: Fehlerbehandlung. HTTP-Fehler der Plattform und Wiederholungsversuche: Fehlerbehandlung.

Aktualität der Daten

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

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 für das Muster mit dem message-Schlüssel.

Caching

Cachen Sie Abfragen zu Luftfahrzeugregister und Flugleistungsdaten aggressiv. Live-ADS-B-Daten dürfen niemals gecacht werden — eine Position aus dem Cache auszuliefern, widerspricht dem Zweck der Echtzeitverfolgung.

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

Live-ADS-B-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-Abruf vollständig vom Rendering-Zyklus — verwenden Sie ein Hintergrundintervall oder einen Datenspeicher.

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

Protokollieren Sie diese Werte bei jeder ADS-B- und Luftfahrzeuganfrage:

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

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 · Live-Tracking · Fehlerbehandlung