[{"sha":"e8788fae83bb19b23baef9c310056223ff1b86e9","filename":"mpt_api_client/resources/webhooks.py","status":"added","additions":100,"deletions":0,"changes":100,"blob_url":"https://github.com/softwareone-platform/mpt-api-python-client/blob/f90c0e9914bb33ef03997ecf35fca8d0371bf486/mpt_api_client%2Fresources%2Fwebhooks.py","raw_url":"https://github.com/softwareone-platform/mpt-api-python-client/raw/f90c0e9914bb33ef03997ecf35fca8d0371bf486/mpt_api_client%2Fresources%2Fwebhooks.py","contents_url":"https://api.github.com/repos/softwareone-platform/mpt-api-python-client/contents/mpt_api_client%2Fresources%2Fwebhooks.py?ref=f90c0e9914bb33ef03997ecf35fca8d0371bf486","patch":"@@ -0,0 +1,100 @@\n+\"\"\"HTTP service for managing MPT webhook subscriptions.\"\"\"\n+\n+from typing import Any, ClassVar\n+\n+import httpx\n+\n+JsonDict = dict[str, Any]\n+EventCache = dict[str, list[str]]\n+\n+\n+class WebhookService:\n+    \"\"\"Service for creating, reading and deleting webhook subscriptions.\"\"\"\n+\n+    event_cache: ClassVar[EventCache] = {}\n+\n+    def __init__(self, base_url: str, token: str, timeout: float = 20.0) -> None:\n+        \"\"\"Initialize the webhook service.\n+\n+        Args:\n+            base_url: Base URL of the MPT API.\n+            token: Bearer token used to authenticate requests.\n+            timeout: Request timeout in seconds.\n+        \"\"\"\n+        self._client = httpx.Client(base_url=base_url, timeout=timeout)\n+        self._token = token\n+\n+    def create_webhook(self, url: str, events: list[str] | None = None) -> JsonDict:\n+        \"\"\"Create a webhook subscription.\n+\n+        Args:\n+            url: Callback URL that will receive webhook deliveries.\n+            events: Event names to subscribe to.\n+\n+        Returns:\n+            The created webhook as returned by the API.\n+\n+        Raises:\n+            MPTError: If the API responds with an error status.\n+        \"\"\"\n+        payload = {\"url\": url, \"events\": events or []}\n+        response = self._client.post(\"/webhooks\", json=payload, headers=self._headers())\n+        created: JsonDict = response.json()\n+        return created\n+\n+    def fetch_webhook(self, webhook_id: str) -> JsonDict:\n+        \"\"\"Retrieve a single webhook by its identifier.\n+\n+        Args:\n+            webhook_id: Identifier of the webhook to retrieve.\n+\n+        Returns:\n+            The webhook as returned by the API.\n+        \"\"\"\n+        response = self._client.get(f\"/webhooks/{webhook_id}\", headers=self._headers())\n+        webhook: JsonDict = response.json()\n+        return webhook\n+\n+    def delete_webhook(self, webhook_id: str) -> bool:\n+        \"\"\"Delete a webhook subscription.\n+\n+        Args:\n+            webhook_id: Identifier of the webhook to delete.\n+\n+        Returns:\n+            True if the webhook was deleted successfully.\n+        \"\"\"\n+        response = self._client.request(\n+            \"DELETE\", f\"/webhooks/{webhook_id}\", headers=self._headers()\n+        )\n+        return bool(response.status_code == httpx.codes.OK)\n+\n+    def list_event_types(self) -> list[str]:\n+        \"\"\"Return the names of all supported webhook event types.\n+\n+        Results are cached after the first successful call.\n+\n+        Returns:\n+            The list of supported event type names.\n+        \"\"\"\n+        if self.event_cache.get(\"events\"):\n+            return self.event_cache[\"events\"]\n+        response = self._client.get(\"/webhooks/events\", headers=self._headers())\n+        names = [event[\"name\"] for event in response.json()]\n+        self.event_cache[\"events\"] = names\n+        return names\n+\n+    def is_active(self, webhook: JsonDict) -> bool:\n+        \"\"\"Return whether a webhook subscription is currently active.\n+\n+        Args:\n+            webhook: A webhook payload returned by the API.\n+\n+        Returns:\n+            True if the webhook status is active.\n+        \"\"\"\n+        return bool(webhook[\"status\"] == \"active\")\n+\n+    def _headers(self) -> dict[str, str]:\n+        \"\"\"Build the authorization headers for a request.\"\"\"\n+        return {\"Authorization\": f\"Bearer {self._token}\"}"}]