Caching & monitoring

Empty-list and zero-alert semantics: Error handling. Platform HTTP errors and retries: Error Handling.

Data freshness

Endpoint	Update cadence	Recommended TTL	Notes
`GET /notams/{icao}`	Continuous	5 min max	NOTAMs are issued, amended, and cancelled at any time
`GET /delays/faa`	Every 2–5 min (ATCSCC)	2–5 min	Nationwide snapshot; ground stops can appear with no advance notice
`GET /delays/faa/{icao}`	Every 2–5 min (ATCSCC)	2–5 min	Airport-specific slice of the same feed

Do not cache longer than the recommended TTL for either endpoint. A ground stop issued while a stale cache entry is served means your users see "no active delays" during an active program.

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

NOTAM_TTL = 300        # 5 minutes
FAA_DELAY_TTL = 120    # 2 minutes


def cached_get(url: str, ttl_seconds: int) -> dict | None:
    entry = _cache.get(url)
    if entry and time.time() < entry[1]:
        return entry[0]

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


def get_notams(icao: str) -> dict | None:
    """Cached NOTAM fetch — 5-minute TTL."""
    return cached_get(f"{BASE}/notams/{icao}", NOTAM_TTL)


def get_faa_delays_nationwide() -> dict | None:
    """Cached nationwide FAA delay snapshot — 2-minute TTL."""
    return cached_get(f"{BASE}/delays/faa", FAA_DELAY_TTL)


# data = get_notams("KJFK")
# delays = get_faa_delays_nationwide()

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

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 and alerting

Log every NOTAM and FAA delay request, and set up active alerts for operationally significant changes:

import logging
import time
import requests

logger = logging.getLogger("alerts")

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"

# Track previous counts to detect sudden changes
_prev_notam_counts: dict[str, int] = {}
_prev_faa_ground_stops: set[str] = set()

WATCHED_AIRPORTS = ["KJFK", "KLAX", "KORD"]


def check_notams_for_airport(icao: str) -> None:
    start = time.monotonic()
    r = requests.get(f"{BASE}/notams/{icao}", headers=HEADERS, timeout=(10, 15))
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "notams",
        "icao": icao,
        "status": r.status_code,
        "elapsed_ms": round(elapsed_ms),
        "quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"),
    }

    if r.status_code == 200:
        data = r.json()
        count = data.get("total", len(data.get("notams", [])))
        log_data["notam_count"] = count
        logger.info("notam_fetch", extra=log_data)

        prev = _prev_notam_counts.get(icao)
        if prev is not None and abs(count - prev) >= 5:
            logger.warning(
                "notam_count_jump",
                extra={"icao": icao, "prev": prev, "current": count},
            )
        _prev_notam_counts[icao] = count

    elif r.status_code == 404:
        logger.info("notam_not_found", extra=log_data)
    else:
        logger.warning("notam_error", extra=log_data)
        r.raise_for_status()


def check_faa_ground_stops() -> None:
    start = time.monotonic()
    r = requests.get(f"{BASE}/delays/faa", headers=HEADERS, timeout=(10, 15))
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "delays/faa",
        "status": r.status_code,
        "elapsed_ms": round(elapsed_ms),
        "quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"),
    }

    if r.status_code == 200:
        data = r.json()
        total = data.get("total_alerts", 0)
        log_data["total_alerts"] = total
        logger.info("faa_delay_fetch", extra=log_data)

        # Alert when a ground stop appears at a watched airport
        current_gs = {
            gs["airport"]
            for gs in data.get("ground_stops", [])
            if gs.get("airport") in WATCHED_AIRPORTS
        }
        new_gs = current_gs - _prev_faa_ground_stops
        if new_gs:
            logger.warning(
                "ground_stop_appeared",
                extra={"airports": sorted(new_gs)},
            )
        _prev_faa_ground_stops.clear()
        _prev_faa_ground_stops.update(current_gs)

    else:
        logger.warning("faa_delay_error", extra=log_data)
        r.raise_for_status()


# for icao in WATCHED_AIRPORTS:
#     check_notams_for_airport(icao)
# check_faa_ground_stops()

Alert on:

notam_count_jump — NOTAM count at a watched airport changes by 5 or more between polls; an unusual surge may indicate airspace restrictions or emergency notices
ground_stop_appeared — a ground stop activates at any airport in your watch list; ground stops can block all departures with minutes of notice
elapsed_ms > 5000 repeatedly — upstream latency issue; consider a timeout and fallback UI
quota_remaining < 200 — close to monthly limit; reduce poll frequency or upgrade plan
status == 429 more than twice in a 5-minute window — your polling rate is too high

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[list, float]] = {}


def get_notams(icao: str) -> list[dict]:
    key = icao.upper()
    hit = _cache.get(key)
    if hit and time.time() < hit[1]:
        return hit[0]
    r = requests.get(f"{BASE}/notams/{key}", headers=HEADERS, timeout=(10, 15))
    r.raise_for_status()
    rows = r.json().get("notams") or []
    _cache[key] = (rows, time.time() + 300)
    return rows


if __name__ == "__main__":
    rows = get_notams("KJFK")
    print(f"KJFK NOTAMs: {len(rows)} (first call)")
    print(f"Cached repeat: {len(get_notams('KJFK'))} NOTAMs")

Related: NOTAMs · FAA delays · Error handling

Data freshness

Caching

Rate limits

Monitoring and alerting

Runnable example

On this page