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

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

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

Тип данных	График обновления	Примечания
METAR	Каждые 30 мин (крупные аэропорты), каждые 60 мин (небольшие станции)	Сводки SPECI выпускаются немедленно при значительном изменении условий
TAF	4 раза в сутки — 00Z, 06Z, 12Z, 18Z	Действует 24 или 30 часов; TAF остаётся актуальным между выпусками
Ветер на высотах (FB)	Фиксированный график продукта FAA	Только США; `valid_time` в ответе указывает срок действия прогноза
PIREP	Пилоты подают донесения непрерывно	Быстро устаревают — используйте короткий период ретроспективы (параметр `hours`)
AIRMET/SIGMET	Выпускаются и обновляются по графику FAA	Отбрасывайте истёкшие на стороне клиента по `start_time` / `end_time`

Поле timestamp в каждом ответе — это момент, когда API получил данные; оно меняется при каждом запросе, даже если само наблюдение не изменилось. Для определения возраста METAR используйте parsed.time (время наблюдения из заголовка сводки). Для TAF не используйте parsed.start_time как показатель свежести сводки — это начало периода действия прогноза. Между четырьмя ежедневными выпусками один и тот же TAF остаётся актуальным; опрашивайте не чаще чем раз в 15–30 минут или после ожидаемого времени выпуска (00Z, 06Z, 12Z, 18Z).

import requests
from datetime import datetime, timezone

HEADERS = {
    "X-RapidAPI-Key": "YOUR_RAPIDAPI_KEY",
    "X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"


def parse_observation_time(iso: str) -> datetime:
    return datetime.fromisoformat(iso.replace("Z", "+00:00"))


def observation_age_minutes(metar: dict) -> float:
    obs_time = parse_observation_time(metar["parsed"]["time"])
    now = datetime.now(timezone.utc)
    return (now - obs_time).total_seconds() / 60


def fetch_metar(icao: str) -> dict:
    r = requests.get(
        f"{BASE}/weather/metar/{icao}",
        headers=HEADERS,
        params={"parsed": "true"},
        timeout=(10, 15),
    )
    r.raise_for_status()
    return r.json()


# age = observation_age_minutes(data)
# if age > 90:
#     print(f"Warning: observation is {age:.0f} minutes old - station may be offline")

Если observation_age_minutes возвращает больше 90 минут, станция, скорее всего, отключена или передаёт данные с перебоями. Данные всё равно можно показывать, но пометьте их как возможно устаревшие.

Кэширование

Ответы METAR и TAF меняются медленно по сравнению с частотой обновления интерфейса. Кэшируйте METAR на 3–5 минут, а TAF — на 15–30 минут, вместо того чтобы запрашивать данные заново при каждой отрисовке.

Эндпоинт	Рекомендуемый TTL	Ключ кэша
`GET /weather/metar/{icao}`	3–5 минут	`metar:{icao}`
`GET /weather/metar/{icao}?parsed=true`	3–5 минут	`metar:{icao}:parsed`
`GET /weather/taf/{icao}`	15–30 минут	`taf:{icao}`
`GET /weather/taf/{icao}?parsed=true`	15–30 минут	`taf:{icao}:parsed`
`GET /weather/winds-aloft?bbox=…`	30–60 минут	`winds:{bbox}:{forecast}:{level}`
`GET /weather/pireps?bbox=…`	5–15 минут	`pirep:{bbox}:{hours}`
`GET /weather/airsigmet?bbox=…`	10–15 минут	`airsig:{bbox}:{type}`

import time
import requests

HEADERS = {
    "X-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()))
    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_metar(icao: str) -> dict | None:
    return cached_get(f"{BASE}/weather/metar/{icao}", {"parsed": "true"}, ttl_seconds=300)


def get_taf(icao: str) -> dict | None:
    return cached_get(f"{BASE}/weather/taf/{icao}", {"parsed": "true"}, ttl_seconds=1800)

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

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

Каждый ответ содержит заголовки rate limit:

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

import requests

HEADERS = {
    "X-RapidAPI-Key": "YOUR_RAPIDAPI_KEY",
    "X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}


def check_quota() -> dict:
    r = requests.get(
        "https://skylink-api.p.rapidapi.com/weather/metar/KJFK",
        headers=HEADERS,
        timeout=(10, 15),
    )
    r.raise_for_status()

    remaining = int(r.headers.get("X-RateLimit-Requests-Remaining", 0))
    limit      = int(r.headers.get("X-RateLimit-Requests-Limit", 0))
    reset_in   = int(r.headers.get("X-RateLimit-Requests-Reset", 0))

    return {
        "remaining": remaining,
        "limit": limit,
        "reset_in": reset_in,
        "low_quota": remaining < 100,
    }


# quota = check_quota()
# print(f"Quota: {quota['remaining']}/{quota['limit']} remaining, resets in {quota['reset_in']}s")
#
# if quota["low_quota"]:
#     print("Warning: approaching quota limit - reduce polling frequency or upgrade plan")

При получении 429 тело ответа имеет вид {"message": "Too many requests"} (формат шлюза RapidAPI). Подождите и повторите попытку — см. вспомогательную функцию повторных попыток на странице «Обработка ошибок» платформы.

Мониторинг и логирование

Логируйте при каждом запросе погоды:

import logging
import time
import requests
from datetime import datetime, timezone

logger = logging.getLogger("weather")
HEADERS = {
    "X-RapidAPI-Key": "YOUR_RAPIDAPI_KEY",
    "X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"


def fetch_metar_logged(icao: str) -> dict | None:
    start = time.monotonic()
    r = requests.get(f"{BASE}/weather/metar/{icao}", headers=HEADERS, params={"parsed": "true"}, timeout=(10, 15))
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "metar",
        "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()
        if "parsed" in data:
            obs_time = datetime.fromisoformat(data["parsed"]["time"].replace("Z", "+00:00"))
            age_min = (datetime.now(timezone.utc) - obs_time).total_seconds() / 60
            log_data["observation_age_min"] = round(age_min, 1)
            log_data["flight_rules"] = data["parsed"]["flight_rules"]
        logger.info("metar_fetch", extra=log_data)
        return data
    elif r.status_code == 404:
        logger.info("metar_not_found", extra=log_data)
        return None
    else:
        logger.warning("metar_error", extra=log_data)
        r.raise_for_status()

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

observation_age_min > 90 — станция, скорее всего, отключена или обновления задерживаются
status = 429 более двух раз за 5 минут — частота опроса слишком высока для вашего тарифа
повторяющиеся elapsed_ms > 5000 — проблема с задержкой у поставщика данных; рассмотрите таймаут и резервный вариант
quota_remaining < 200 — вы близки к месячному лимиту

См. также: METAR · TAF · Ветер на высотах · PIREP · AIRMET/SIGMET · Обработка ошибок · Сценарии для аэропортов · Сценарии на маршруте

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

Кэширование

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

Мониторинг и логирование

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