# Кэширование и мониторинг Паттерны для продакшна при работе с данными о полётах — кэширование, ограничения частоты запросов и мониторинг. Обработка ошибок и особенности полей: [Обработка ошибок](/docs/v31/flight-operations-integration-errors). HTTP-ошибки платформы и повторные попытки: [Обработка ошибок](/docs/errors). ### Актуальность данных [#data-freshness] Данные о полётах варьируются от почти реального времени (статус рейса) до статических (расстояние). Подбирайте TTL под частоту обновления: избыточное кэширование живых данных ухудшает UX, недостаточное кэширование статических данных впустую расходует квоту. | Эндпоинт | Рекомендуемый TTL | Примечания | | ------------------------------- | ------------------------------------ | ------------------------------------------------------------------------------------------------- | | `GET /flight_status/{callsign}` | 30–60 секунд | Отслеживание в реальном времени; позиция и статус меняются каждую минуту | | `GET /schedules` | 2–5 минут | Расписания стабильны в пределах сессии, но обновляются при нерегулярных операциях | | `GET /distance` | 24 часа | Расстояние по ортодромии для фиксированного маршрута неизменно | | `GET /briefing` | Не кэшировать (или максимум 5 минут) | Генерируется ИИ для каждого запроса; запрашивайте заново, если пользователь явно обновляет данные | ### Задержка брифинга [#briefing-latency] Эндпоинт брифинга обращается к бэкенду ИИ-инференса ([IBM Granite](https://www.ibm.com/granite)) для генерации предвылетной сводки. Рассчитывайте на время ответа **60–90 секунд**. Учитывайте это при проектировании: * Сразу показывайте индикатор загрузки, когда пользователь запрашивает брифинг. * Используйте `timeout=(10, 90)` — таймаут подключения может оставаться 10 с, но таймаут чтения необходимо увеличить. * Выполняйте одну повторную попытку при `502`/`503`, прежде чем показывать ошибку. ### Кэширование [#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]] = {} 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]: data = cached_get( f"{BASE}/schedules", {"airport": icao, "direction": direction}, 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, ) ``` В продакшне замените `_cache` на Redis или Memcached, чтобы кэш был общим для всех экземпляров. ### Ограничения частоты запросов [#rate-limits] Каждый ответ содержит заголовки ограничения частоты запросов. Уровни квот и паттерн повторных попыток при 429 описаны на странице [Обработка ошибок](/docs/errors). | Заголовок | Значение | | -------------------------------- | ------------------------------------- | | `X-RateLimit-Requests-Limit` | Месячная квота запросов вашего тарифа | | `X-RateLimit-Requests-Remaining` | Оставшиеся запросы в этом месяце | | `X-RateLimit-Requests-Reset` | Секунды до сброса квоты | Опрос статуса рейса с TTL 45 секунд обходится примерно в 2 запроса в минуту на каждый отслеживаемый рейс. Подбирайте тариф соответственно: для табло вылетов на 10 рейсов это \~2880 запросов в день. ### Готовый к запуску пример [#runnable-example] Получает расстояние KJFK–KLAX с кэшем на 24 часа и выводит попадание/промах кэша при каждом вызове. ```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]] = {} 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] Логируйте эти поля для каждого запроса к данным о полётах: ```python 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() ``` **Настройте оповещения при:** * `elapsed_ms > 5000` для статуса рейса — задержка на стороне upstream-сервиса; сократите частоту опроса или добавьте резервный вариант * `elapsed_ms > 95000` для брифинга — истекло время генерации; предложите пользователю повторить попытку * `status = 429` — частота опроса превышает квоту тарифа; увеличьте TTL или перейдите на более высокий тариф * `quota_remaining < 500` — приближение к месячному лимиту; пересмотрите частоту опроса *** См. также: [Статус рейса](/docs/v31/flight-status-integration) · [Расписания](/docs/v31/schedules-integration) · [Расстояние](/docs/v31/distance-integration) · [Брифинг по полёту](/docs/v31/flight-briefing-integration) · [Обработка ошибок](/docs/v31/flight-operations-integration-errors)