Added webhook management functions

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

@@ -18,7 +18,6 @@
     LLMWhispererClientException: Exception raised for errors in the LLMWhispererClient.
 """
 
-import copy
 import json
 import logging
 import os
@@ -504,9 +503,44 @@ def register_webhook(self, url: str, auth_token: str, webhook_name: str) -> dict
             "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("POST", url, headers=headersx, json=data)
+        req = requests.Request("POST", url, headers=self.headers, json=data)
+        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"
+        req = requests.Request("PUT", url, headers=self.headers, json=data)
         prepared = req.prepare()
         s = requests.Session()
         response = s.send(prepared, timeout=self.api_timeout)
@@ -547,6 +581,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"

tests/unit/client_v2_test.py

@@ -1,15 +1,15 @@
 WEBHOOK_URL = "http://test-webhook.com/callback"
 AUTH_TOKEN = "dummy-auth-token"
 WEBHOOK_NAME = "test_webhook"
-WEBHOOK_RESPONSE = {"status": "success", "message": "Webhook registered successfully"}
+WEBHOOK_RESPONSE = {"message": "Webhook registered successfully"}
 WHISPER_RESPONSE = {"status_code": 200, "extraction": {"result_text": "Test result"}}
 
 
 def test_register_webhook(mocker, client_v2):
     mock_send = mocker.patch("requests.Session.send")
     mock_response = mocker.MagicMock()
-    mock_response.status_code = 200
-    mock_response.text = '{"status": "success", "message": "Webhook registered successfully"}'  # noqa: E501
+    mock_response.status_code = 201
+    mock_response.text = '{"message": "Webhook registered successfully"}'  # noqa: E501
     mock_send.return_value = mock_response
 
     response = client_v2.register_webhook(WEBHOOK_URL, AUTH_TOKEN, WEBHOOK_NAME)