Webhook-Abonnements

import osimport requestsBASE = "https://data.skylinkapi.com/v3.1"HEADERS = {    "x-api-key": os.getenv("SKYLINK_API_KEY") or os.getenv("SKYLINK_API_KEY", "YOUR_API_KEY")}def list_webhook_event_types() -> list[str]:    """Return supported webhook event type strings."""    r = requests.get(f"{BASE}/webhooks/events", headers=HEADERS, timeout=(10, 15))    r.raise_for_status()    return r.json()["event_types"]if __name__ == "__main__":    import json    print(json.dumps(list_webhook_event_types(), indent=2))

"""Webhook subscription models."""from __future__ import annotationsfrom dataclasses import dataclass@dataclassclass WebhookCreateRequest:    url: str    event_types: list[str]    filters: dict[str, str]@dataclassclass WebhookSubscription:    id: str    url: str    event_types: list[str]    filters: dict[str, str]    active: bool@dataclassclass WebhookListResponse:    count: int    webhooks: list[WebhookSubscription]@dataclassclass WebhookEventTypesResponse:    event_types: list[str]@dataclassclass WebhookDelivery:    """Inbound POST body SkyLink sends to your callback URL."""    event_type: str    timestamp: str    webhook_id: str    flight_number: str    airline: str    status: str    departure: dict[str, str]    arrival: dict[str, str]

import osimport requestsBASE = "https://data.skylinkapi.com/v3.1"HEADERS = {    "x-api-key": os.getenv("SKYLINK_API_KEY", "YOUR_API_KEY")}def list_webhook_event_types() -> list[str]:    """Return supported webhook event type strings."""    r = requests.get(f"{BASE}/webhooks/events", headers=HEADERS, timeout=(10, 15))    r.raise_for_status()    return r.json()["event_types"]def list_webhooks() -> dict:    """List webhook subscriptions for the current API key."""    r = requests.get(f"{BASE}/webhooks", headers=HEADERS, timeout=(10, 15))    r.raise_for_status()    return r.json()def create_webhook(*, url: str, event_types: list[str], flight_number: str) -> dict:    """Create a webhook subscription for a single flight."""    body = {        "url": url,        "event_types": event_types,        "filters": {"flight_number": flight_number.upper().replace(" ", "")},    }    r = requests.post(f"{BASE}/webhooks", headers=HEADERS, json=body, timeout=(10, 20))    r.raise_for_status()    return r.json()def set_webhook_active(webhook_id: str, *, active: bool) -> dict:    """Enable or disable a webhook without deleting it."""    r = requests.patch(        f"{BASE}/webhooks/{webhook_id}",        headers=HEADERS,        json={"active": active},        timeout=(10, 15),    )    r.raise_for_status()    return r.json()def delete_webhook(webhook_id: str) -> None:    """Permanently delete a webhook subscription."""    r = requests.delete(f"{BASE}/webhooks/{webhook_id}", headers=HEADERS, timeout=(10, 15))    r.raise_for_status()if __name__ == "__main__":    import json    events = list_webhook_event_types()    print(json.dumps(events, indent=2))