Added webhook management functions

Author: johnyrahul

src/unstract/llmwhisperer/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "2.2.1"
+__version__ = "2.3.0"
 
 from .client import LLMWhispererClient  # noqa: F401
 from .client_v2 import LLMWhispererClientV2  # noqa: F401

src/unstract/llmwhisperer/client_v2.py

@@ -510,6 +510,45 @@ def register_webhook(self, url: str, auth_token: str, webhook_name: str) -> dict
         prepared = req.prepare()
         s = requests.Session()
         response = s.send(prepared, timeout=self.api_timeout)
+        if response.status_code != 201:
+            err = json.loads(response.text)
+            err["status_code"] = response.status_code
+            raise LLMWhispererClientException(err)
+        return json.loads(response.text)
+
+    def update_webhook_details(self, webhook_name: str, url: str, auth_token: str) -> dict:
+        """Updates the details of a webhook from the LLMWhisperer API.
+
+        This method sends a PUT request to the '/whisper-manage-callback' endpoint of the LLMWhisperer API.
+        The response is a JSON object containing the status of the webhook update.
+
+        Refer to https://docs.unstract.com/llm_whisperer/apis/
+
+        Args:
+            webhook_name (str): The name of the webhook.
+            url (str): The URL of the webhook.
+            auth_token (str): The authentication token for the webhook.
+
+        Returns:
+            dict: A dictionary containing the status code and the response from the API.
+
+        Raises:
+            LLMWhispererClientException: If the API request fails, it raises an exception with
+                                            the error message and status code returned by the API.
+        """
+
+        data = {
+            "url": url,
+            "auth_token": auth_token,
+            "webhook_name": webhook_name,
+        }
+        url = f"{self.base_url}/whisper-manage-callback"
+        headersx = copy.deepcopy(self.headers)
+        headersx["Content-Type"] = "application/json"
+        req = requests.Request("PUT", url, headers=headersx, json=data)
+        prepared = req.prepare()
+        s = requests.Session()
+        response = s.send(prepared, timeout=self.api_timeout)
         if response.status_code != 200:
             err = json.loads(response.text)
             err["status_code"] = response.status_code
@@ -547,6 +586,37 @@ def get_webhook_details(self, webhook_name: str) -> dict:
             raise LLMWhispererClientException(err)
         return json.loads(response.text)
 
+    def delete_webhook(self, webhook_name: str) -> dict:
+        """Deletes a webhook from the LLMWhisperer API.
+
+        This method sends a DELETE request to the '/whisper-manage-callback' endpoint of the LLMWhisperer API.
+        The response is a JSON object containing the status of the webhook deletion.
+
+        Refer to https://docs.unstract.com/llm_whisperer/apis/
+
+        Args:
+            webhook_name (str): The name of the webhook.
+
+        Returns:
+            dict: A dictionary containing the status code and the response from the API.
+
+        Raises:
+            LLMWhispererClientException: If the API request fails, it raises an exception with
+                                            the error message and status code returned by the API.
+        """
+
+        url = f"{self.base_url}/whisper-manage-callback"
+        params = {"webhook_name": webhook_name}
+        req = requests.Request("DELETE", url, headers=self.headers, params=params)
+        prepared = req.prepare()
+        s = requests.Session()
+        response = s.send(prepared, timeout=self.api_timeout)
+        if response.status_code != 200:
+            err = json.loads(response.text)
+            err["status_code"] = response.status_code
+            raise LLMWhispererClientException(err)
+        return json.loads(response.text)
+
     def get_highlight_rect(
         self,
         line_metadata: list[int],

tests/integration/client_v2_test.py

@@ -5,6 +5,8 @@
 
 import pytest
 
+from unstract.llmwhisperer.client_v2 import LLMWhispererClientException
+
 logger = logging.getLogger(__name__)
 
 
@@ -168,6 +170,63 @@ def test_whisper_v2_url_in_post(client_v2, data_dir, output_mode, mode, url, inp
     verify_usage(usage_before, usage_after, page_count, mode)
 
 
+@pytest.mark.parametrize(
+    "url,token,webhook_name",
+    [
+        (
+            "https://webhook.site/b76ecc5f-8320-4410-b24f-66525d2c92cb",
+            "",
+            "client_v2_test",
+        ),
+    ],
+)
+def test_webhook(client_v2, url, token, webhook_name):
+    """Tests the registration, retrieval, update, and deletion of a webhook.
+
+    This test method performs the following steps:
+    1. Registers a new webhook with the provided URL, token, and webhook name.
+    2. Retrieves the details of the registered webhook and verifies the URL, token, and webhook name.
+    3. Updates the webhook details with a new token.
+    4. Deletes the webhook and verifies the deletion.
+
+    Args:
+        client_v2 (LLMWhispererClientV2): The client instance for making API requests.
+        url (str): The URL of the webhook.
+        token (str): The authentication token for the webhook.
+        webhook_name (str): The name of the webhook.
+
+    Returns:
+        None
+    """
+    result = client_v2.register_webhook(url, token, webhook_name)
+    assert isinstance(result, dict)
+    assert result["message"] == "Webhook created successfully"
+
+    result = client_v2.get_webhook_details(webhook_name)
+    assert isinstance(result, dict)
+    assert result["url"] == url
+    assert result["auth_token"] == token
+    assert result["webhook_name"] == webhook_name
+
+    result = client_v2.update_webhook_details(webhook_name, url, "new_token")
+    assert isinstance(result, dict)
+    assert result["message"] == "Webhook updated successfully"
+
+    result = client_v2.get_webhook_details(webhook_name)
+    assert isinstance(result, dict)
+    assert result["auth_token"] == "new_token"
+
+    result = client_v2.delete_webhook(webhook_name)
+    assert isinstance(result, dict)
+    assert result["message"] == "Webhook deleted successfully"
+
+    try:
+        client_v2.get_webhook_details(webhook_name)
+    except LLMWhispererClientException as e:
+        assert e.error_message()["message"] == "Webhook details not found"
+        assert e.error_message()["status_code"] == 404
+
+
 def assert_error_message(whisper_result):
     assert isinstance(whisper_result, dict)
     assert whisper_result["status"] == "error"