Caching & monitoring

Error handling and field quirks: Error handling. Platform HTTP errors and retries: Error Handling.

Data freshness

Flight operations data ranges from near-real-time (flight status) to static (distance). Match your TTL to the update cadence — over-caching live data degrades UX, under-caching static data wastes quota.

Endpoint	Recommended TTL	Notes
`GET /flight_status/{callsign}`	30–60 seconds	Live tracking; position and status change every minute
`GET /schedules/departures`, `/schedules/arrivals`	2–5 minutes	Schedules are stable within a session but update during irregular ops
`GET /distance`	24 hours	Great-circle distance is static for a fixed route
`GET /briefing`	Do not cache (or max 5 minutes)	Generated per-request by AI; re-fetch if user explicitly refreshes

Briefing latency

The briefing endpoint calls an AI inference backend (IBM Granite) to generate a pre-departure summary. Expect 60–90 seconds of response time. Design for this:

Show a loading indicator immediately when the user requests a briefing.
Use timeout=(10, 90) — the connect timeout can stay at 10s, but the read timeout must be extended.
Retry once on 502/503 before surfacing an error.

Caching

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]:
    if direction not in ("departures", "arrivals"):
        raise ValueError("direction must be 'departures' or 'arrivals'")
    data = cached_get(
        f"{BASE}/schedules/{direction}",
        {"icao": icao},
        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,
    )

For production, replace _cache with Redis or Memcached so the cache is shared across instances.

Rate limits

Every response includes rate-limit headers. See Error Handling for quota tiers and the 429 retry pattern.

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

Flight status polling at 45-second TTL costs roughly 2 requests/minute per tracked flight. Scale your plan accordingly — for a 10-flight departure board, that is ~2,880 requests/day.

Runnable example

Fetches KJFK–KLAX distance with a 24-hour TTL cache, printing hit/miss on each call.

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

Log these fields on every flight operations request:

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()

Alert on:

elapsed_ms > 5000 on flight status — upstream latency; reduce polling or add a fallback
elapsed_ms > 95000 on briefing — generation timed out; surface a retry option to the user
status = 429 — polling rate exceeds plan quota; increase TTL or upgrade
quota_remaining < 500 — approaching monthly limit; review polling frequency