# Кэширование и мониторинг Шаблоны работы с NOTAM и задержками FAA в продакшне — частота опроса, кэширование и мониторинг. Семантика пустых списков и нулевого числа оповещений: [Обработка ошибок](/docs/v31/alerts-integration-errors). HTTP-ошибки платформы и повторные попытки: [Обработка ошибок](/docs/errors). ### Актуальность данных [#data-freshness] | Эндпоинт | Частота обновления | Рекомендуемый TTL | Примечания | | ------------------------ | ----------------------- | ------------------ | ------------------------------------------------------------------------------------------------------- | | `GET /notams/{icao}` | Непрерывно | **не более 5 мин** | NOTAM выпускаются, изменяются и отменяются в любой момент | | `GET /delays/faa` | Каждые 2–5 мин (ATCSCC) | 2–5 мин | Снимок по всей стране; запреты на вылет (ground stop) могут появляться без предварительного уведомления | | `GET /delays/faa/{icao}` | Каждые 2–5 мин (ATCSCC) | 2–5 мин | Срез того же потока по конкретному аэропорту | **Не кэшируйте дольше рекомендуемого TTL ни для одного из эндпоинтов.** Если запрет на вылет введён, пока из кэша отдаётся устаревшая запись, ваши пользователи увидят «нет действующих задержек» во время действующей программы. ### Кэширование [#caching] ```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]] = {} 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() ``` В продакшне замените кэш-словарь на общее хранилище (Redis, Memcached), чтобы кэш был согласован между экземплярами и переживал перезапуски. ### Ограничения частоты запросов [#rate-limits] Каждый ответ содержит заголовки ограничения частоты запросов: | Заголовок | Значение | | -------------------------------- | --------------------------------------- | | `X-RateLimit-Requests-Limit` | Месячная квота запросов вашего тарифа | | `X-RateLimit-Requests-Remaining` | Сколько запросов осталось в этом месяце | | `X-RateLimit-Requests-Reset` | Секунд до сброса квоты | Квоты тарифов: | Тариф | Запросов в месяц | | ----- | ---------------- | | Free | 1000 | | Basic | 5000 | | Pro | 50 000 | | Ultra | 200 000 | | Mega | 600 000 | ### Мониторинг и оповещения [#monitoring-and-alerting] Записывайте в лог каждый запрос [NOTAM](https://www.icao.int/safety/pages/notam-service-profiles.aspx) и задержек FAA и настройте активные оповещения об эксплуатационно значимых изменениях: ```python 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() ``` **Настройте оповещения на:** * `notam_count_jump` — число [NOTAM](https://www.icao.int/safety/pages/notam-service-profiles.aspx) в отслеживаемом аэропорту изменилось на 5 или более между опросами; необычный всплеск может указывать на ограничения воздушного пространства или экстренные уведомления * `ground_stop_appeared` — в любом аэропорту из вашего списка отслеживания введён запрет на вылет (ground stop); такие запреты могут остановить все вылеты с уведомлением всего за несколько минут * `elapsed_ms > 5000` повторно — проблема с задержками на стороне внешнего сервиса; предусмотрите тайм-аут и резервный интерфейс * `quota_remaining < 200` — вы близки к месячному лимиту; уменьшите частоту опроса или перейдите на более высокий тариф * `status == 429` более двух раз за 5 минут — частота опроса слишком высока ### Готовый к запуску пример [#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[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") ``` *** См. также: [NOTAM](/docs/v31/notams-integration) · [Задержки FAA](/docs/v31/faa-delays-integration) · [Обработка ошибок](/docs/v31/alerts-integration-errors)