Кэширование и мониторинг

Обработка ошибок и особенности полей: Обработка ошибок. HTTP-ошибки платформы и повторные попытки: Обработка ошибок.

Актуальность данных

Данные о полётах варьируются от почти реального времени (статус рейса) до статических (расстояние). Подбирайте TTL под частоту обновления: избыточное кэширование живых данных ухудшает UX, недостаточное кэширование статических данных впустую расходует квоту.

Эндпоинт	Рекомендуемый TTL	Примечания
`GET /flight_status/{callsign}`	30–60 секунд	Отслеживание в реальном времени; позиция и статус меняются каждую минуту
`GET /schedules`	2–5 минут	Расписания стабильны в пределах сессии, но обновляются при нерегулярных операциях
`GET /distance`	24 часа	Расстояние по ортодромии для фиксированного маршрута неизменно
`GET /briefing`	Не кэшировать (или максимум 5 минут)	Генерируется ИИ для каждого запроса; запрашивайте заново, если пользователь явно обновляет данные

Задержка брифинга

Эндпоинт брифинга обращается к бэкенду ИИ-инференса (IBM Granite) для генерации предвылетной сводки. Рассчитывайте на время ответа 60–90 секунд. Учитывайте это при проектировании:

Сразу показывайте индикатор загрузки, когда пользователь запрашивает брифинг.
Используйте timeout=(10, 90) — таймаут подключения может оставаться 10 с, но таймаут чтения необходимо увеличить.
Выполняйте одну повторную попытку при 502/503, прежде чем показывать ошибку.

Кэширование

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, чтобы кэш был общим для всех экземпляров.

Ограничения частоты запросов

Каждый ответ содержит заголовки ограничения частоты запросов. Уровни квот и паттерн повторных попыток при 429 описаны на странице Обработка ошибок.

Заголовок	Значение
`X-RateLimit-Requests-Limit`	Месячная квота запросов вашего тарифа
`X-RateLimit-Requests-Remaining`	Оставшиеся запросы в этом месяце
`X-RateLimit-Requests-Reset`	Секунды до сброса квоты

Опрос статуса рейса с TTL 45 секунд обходится примерно в 2 запроса в минуту на каждый отслеживаемый рейс. Подбирайте тариф соответственно: для табло вылетов на 10 рейсов это ~2880 запросов в день.

Готовый к запуску пример

Получает расстояние KJFK–KLAX с кэшем на 24 часа и выводит попадание/промах кэша при каждом вызове.

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

Мониторинг

Логируйте эти поля для каждого запроса к данным о полётах:

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 — приближение к месячному лимиту; пересмотрите частоту опроса

См. также: Статус рейса · Расписания · Расстояние · Брифинг по полёту · Обработка ошибок