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

Резервная обработка 404/422 и ограничений по тарифу: Обработка ошибок. HTTP-ошибки платформы и повторные попытки: Обработка ошибок.

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

Эндпоинт	Частота обновления	Рекомендуемый TTL	Примечания
`GET /aircraft/registration/{reg}`	Реестр обновляется нечасто	24 ч	Бортовые номера и эксплуатанты меняются редко
`GET /aircraft/performance/{type}`	Статические характеристики	7 дн.	Значения характеристик фиксированы для каждого указателя типа
`GET /adsb/aircraft`	Каждые несколько секунд	Никогда не кэшировать	Данные о местоположении устаревают к следующей отрисовке
`GET /adsb/aircraft/statistics`	Каждые несколько секунд	Никогда не кэшировать	Общие по потоку счётчики постоянно меняются
`GET /ultra/history/flights`	Архив только с добавлением записей	5–15 мин	Новые рейсы появляются после посадки, а не в реальном времени
`GET /ultra/history/airport/{icao}/traffic`	Архив только с добавлением записей	5–15 мин	Тот же архив; результаты запросов по окну стабилизируются в течение нескольких минут

Уровни тарифов и префиксы путей

Доступ к историческим эндпоинтам зависит от тарифа. Префикс пути определяет максимально доступное окно хранения данных:

Префикс пути	Макс. окно	Минимальный тариф
`/ultra/history/...`	90 дней	Pro
`/mega/history/...`	365 дней	Ultra

Ответ 401 или 403 на любом из префиксов означает, что тариф ключа ниже требуемого порога. Паттерн работы с ключом message описан в разделе Обработка ошибок.

Кэширование

Активно кэшируйте запросы к реестру воздушных судов и к характеристикам. Данные ADS-B в реальном времени никогда не должны кэшироваться — выдача местоположения из кэша лишает смысла отслеживание в реальном времени.

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:
    key = url + str(sorted(params.items()))
    entry = _cache.get(key)
    if entry and time.time() < entry[1]:
        return entry[0]

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


def get_aircraft(registration: str) -> dict | None:
    """Cache registration lookups for 24 hours."""
    return cached_get(
        f"{BASE}/aircraft/registration/{registration}",
        {},
        ttl_seconds=86_400,
    )


def get_performance(icao_type: str) -> dict | None:
    """Cache performance specs for 7 days."""
    return cached_get(
        f"{BASE}/aircraft/performance/{icao_type}",
        {},
        ttl_seconds=604_800,
    )


def fetch_live_adsb(lat: float, lon: float, radius: int) -> list[dict]:
    # Do NOT cache — always fetch fresh positions.
    r = requests.get(
        f"{BASE}/adsb/aircraft",
        headers=HEADERS,
        params={"lat": lat, "lon": lon, "radius": radius, "limit": 50},
        timeout=(10, 15),
    )
    r.raise_for_status()
    return r.json().get("aircraft", [])


# ac = get_aircraft("N636JB")
# if ac and ac.get("found"):
#     perf = get_performance(ac["aircraft"]["icao_type"])
#
# aircraft = fetch_live_adsb(40.6413, -73.7781, 80)  # always live

В продакшне замените кэш на основе словаря общим хранилищем (Redis, Memcached), чтобы кэш сохранялся при перезапусках и был общим для всех экземпляров.

Сценарий использования	Рекомендуемый интервал опроса
Движущаяся карта	5–15 секунд
Виджет информирования о воздушной обстановке	15–30 секунд
Счётчик на дашборде	60 секунд

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

Каждый ответ содержит заголовки ограничения частоты запросов:

Заголовок	Значение
`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

Мониторинг

Логируйте следующие данные для каждого запроса ADS-B и запроса о воздушных судах:

import logging
import time
import requests

logger = logging.getLogger("aircraft")

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_live_adsb_logged(lat: float, lon: float, radius: int) -> list[dict]:
    start = time.monotonic()
    r = requests.get(
        f"{BASE}/adsb/aircraft",
        headers=HEADERS,
        params={"lat": lat, "lon": lon, "radius": radius, "limit": 100},
        timeout=(10, 15),
    )
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "adsb/aircraft",
        "lat": lat,
        "lon": lon,
        "radius": radius,
        "status": r.status_code,
        "elapsed_ms": round(elapsed_ms),
        "quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"),
    }

    if r.status_code == 200:
        aircraft = r.json().get("aircraft", [])
        log_data["aircraft_count"] = len(aircraft)
        logger.info("adsb_fetch", extra=log_data)
        return aircraft

    logger.warning("adsb_error", extra=log_data)
    r.raise_for_status()
    return []

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

aircraft_count == 0 более трёх опросов подряд для одной и той же зоны — это может указывать на сбой потока, а не на действительно пустое воздушное пространство
elapsed_ms > 5000 многократно — проблема с задержкой на стороне источника данных
quota_remaining < 200 — приближение к месячному лимиту; снизьте частоту опроса или перейдите на более высокий тариф
status == 429 более двух раз за 5 минут — интервал опроса слишком агрессивен для вашего тарифа

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

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_aircraft_cached(registration: str) -> dict | None:
    key = registration.upper()
    hit = _cache.get(key)
    if hit and time.time() < hit[1]:
        return hit[0]
    r = requests.get(
        f"{BASE}/aircraft/registration/{key}",
        headers=HEADERS,
        timeout=(10, 15),
    )
    if r.status_code == 404:
        return None
    r.raise_for_status()
    data = r.json()
    _cache[key] = (data, time.time() + 86_400)
    return data


if __name__ == "__main__":
    ac = get_aircraft_cached("N636JB")
    found = ac and ac.get("found")
    print(f"N636JB lookup: {'found' if found else 'not found'}")
    ac2 = get_aircraft_cached("N636JB")
    print("Second call served from 24 h cache:", ac2 is ac)

См. также: Поиск воздушных судов · Отслеживание в реальном времени · Обработка ошибок