Caching & monitoring

404/422 fallbacks and plan-gate handling: Error handling. Platform HTTP errors and retries: Error Handling.

Data freshness

Endpoint	Update cadence	Recommended TTL	Notes
`GET /aircraft/registration/{reg}`	Registry updates infrequently	24 h	Tail numbers and operators change rarely
`GET /adsb/aircraft`	Every few seconds	Never cache	Position data is stale on the next render
`GET /adsb/aircraft/statistics`	Every few seconds	Never cache	Feed-wide counts shift continuously

Caching

Cache aircraft registry lookups aggressively. Live ADS-B must never be cached — serving a position from cache defeats the purpose of real-time tracking.

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 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")
# aircraft = fetch_live_adsb(40.6413, -73.7781, 80)  # always live

For production, replace the dict cache with a shared store (Redis, Memcached) so the cache persists across restarts and is shared across instances.

ADS-B polling guidance

Live ADS-B data refreshes every few seconds on the feed side. Reasonable client polling intervals:

Use case	Recommended poll interval
Moving-map display	5–15 seconds
Traffic awareness widget	15–30 seconds
Dashboard count badge	60 seconds

Do not poll on every render. If your component re-renders on mouse move or scroll events, decouple the ADS-B fetch from the render cycle entirely — use a background interval or a data store.

Rate limits

Every response includes rate-limit headers:

Header	Value
`X-RateLimit-Requests-Limit`	Your plan's monthly request quota
`X-RateLimit-Requests-Remaining`	Requests left this month
`X-RateLimit-Requests-Reset`	Seconds until the quota resets

Plan quotas:

Plan	Requests/month
Free	1,000
Basic	5,000
Pro	50,000
Ultra	200,000
Mega	600,000

Monitoring

Log these on every ADS-B and aircraft request:

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 []

Alert on:

aircraft_count == 0 more than three consecutive polls for the same area — may indicate a feed outage rather than genuinely empty airspace
elapsed_ms > 5000 repeatedly — upstream latency issue
quota_remaining < 200 — approaching monthly limit; reduce polling frequency or upgrade plan
status == 429 more than twice in a 5-minute window — your poll interval is too aggressive for your plan

Runnable example

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)