Schedules

"""Schedules response models for /schedules/departures and /schedules/arrivals."""from __future__ import annotationsfrom dataclasses import dataclass@dataclassclass SchedulesRequest:    """Query input shared by both schedules endpoints."""    icao: str | None = None    iata: str | None = None    date: str | None = None    time: str | None = None    ts: int | None = None@dataclassclass ScheduleFlight:    Time: str    Date: str    IATA: str    Flight: str    Airline: str    Status: str    Destination: str | None = None    Origin: str | None = None@dataclassclass SchedulesResponse:    iata: str    direction: str    airport_code: str    flights: list[ScheduleFlight]    total_flights: int    pages_fetched: int

import osimport requestsBASE = "https://data.skylinkapi.com/v3"HEADERS = {    "x-api-key": os.getenv("SKYLINK_API_KEY", "YOUR_API_KEY")}def fetch_schedules(direction: str, *, icao: str | None = None, iata: str | None = None) -> dict:    """Fetch airport schedules for arrivals or departures."""    if bool(icao) == bool(iata):        raise ValueError("Provide exactly one of icao or iata")    if direction not in {"arrivals", "departures"}:        raise ValueError("direction must be arrivals or departures")    code = icao.upper() if icao else (iata or "").upper()    params = {"icao": code} if icao else {"iata": code}    r = requests.get(f"{BASE}/schedules/{direction}", headers=HEADERS, params=params, timeout=(10, 20))    r.raise_for_status()    return r.json()if __name__ == "__main__":    import json    data = fetch_schedules("departures", icao="KJFK")    flights = data.get("flights") or []    print(json.dumps({"count": len(flights), "sample": flights[:3]}, indent=2, default=str))