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

Резервные сценарии при 404 и ошибки валидации для аэропортов: Обработка ошибок. HTTP-ошибки платформы и повторные попытки: Обработка ошибок.

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

Записи об аэропортах, данные радионавигационных средств и поисковые индексы обновляются периодически, а не в реальном времени. Кэшируйте агрессивно: метаданные аэропортов редко меняются изо дня в день.

Эндпоинт	Рекомендуемый TTL	Примечания
`GET /airports/search`	24 часа	Статические метаданные: ВПП, частоты, превышение
`GET /airports/search/text`	1 час	Поисковый индекс стабилен; в пределах сессии результаты не изменятся
`GET /navaids`	6 часов	Данные радионавигационных средств обновляются по графику аэронавигационных циклов
`GET /airports/search/location`	15 минут	Результаты поиска ближайшего аэропорта зависят от местоположения пользователя; повторяйте запрос при изменении координат

Кэширование

Для интеграций с небольшим объёмом запросов используйте простой TTL-кэш внутри процесса. В развёртываниях с несколькими экземплярами замените словарь на Redis или Memcached, чтобы кэш был общим и переживал перезапуски.

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_airport(icao: str) -> dict | None:
    return cached_get(f"{BASE}/airports/search", {"icao": icao}, ttl_seconds=86400)


def search_airports(query: str) -> list[dict]:
    data = cached_get(f"{BASE}/airports/search/text", {"q": query}, ttl_seconds=3600)
    return (data or {}).get("airports") or []


def get_navaids(icao: str) -> list[dict]:
    data = cached_get(f"{BASE}/navaids", {"airport": icao}, ttl_seconds=21600)
    return (data or {}).get("navaids") or []

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

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

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

Поскольку у данных аэропортов длинный TTL, интеграция с правильно настроенным кэшем расходует очень мало запросов за сессию. Один ответ /airports/search для заданного ICAO-кода можно спокойно использовать повторно до конца дня.

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

Запрашивает KJFK с кэшем на 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_airport(icao: str) -> dict | None:
    url = f"{BASE}/airports/search"
    key = f"airport:{icao}"
    cached = _cache.get(key)
    if cached and time.time() < cached[1]:
        print(f"  cache HIT  — {icao}")
        return cached[0]

    print(f"  cache MISS — {icao}, fetching...")
    r = requests.get(url, headers=HEADERS, params={"icao": icao}, timeout=(10, 15))
    if r.status_code == 404:
        return None
    r.raise_for_status()
    data = r.json()
    _cache[key] = (data, time.time() + 86400)
    return data


if __name__ == "__main__":
    airport = get_airport("KJFK")
    if airport:
        print(f"  {airport.get('name')} — elevation {airport.get('elevation_ft')} ft")

    # Second call — served from cache
    airport = get_airport("KJFK")
    if airport:
        print(f"  {airport.get('name')} — served from cache")

Мониторинг

Записывайте в лог эти поля при каждом запросе к API аэропортов:

import logging
import time
import requests

logger = logging.getLogger("airport")
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_airport_logged(icao: str) -> dict | None:
    start = time.monotonic()
    r = requests.get(
        f"{BASE}/airports/search",
        headers=HEADERS,
        params={"icao": icao},
        timeout=(10, 15),
    )
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "airports/search",
        "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:
        logger.info("airport_fetch", extra=log_data)
        return r.json()
    elif r.status_code == 404:
        logger.info("airport_not_found", extra=log_data)
        return None
    else:
        logger.warning("airport_error", extra=log_data)
        r.raise_for_status()

Настройте оповещения на:

status = 429 — частота опроса превышает возможности вашего тарифа; увеличьте TTL или перейдите на более высокий тариф
elapsed_ms > 5000 повторно — задержки на стороне внешнего сервиса; предусмотрите тайм-аут и резервный сценарий
quota_remaining < 200 — вы приближаетесь к месячному лимиту

См. также: Аэропорты · Поиск аэропортов · Радионавигационные средства · Обработка ошибок

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

Кэширование

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

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

Мониторинг

На этой странице