implemented chat completion and chat workspace methods with docstrings and tests

EazyAl · EazyAl · commit 557700490940 · 2025-08-07T17:14:13.000+02:00
diff --git a/meilisearch/_httprequests.py b/meilisearch/_httprequests.py
@@ -2,7 +2,7 @@
 
 import json
 from functools import lru_cache
-from typing import Any, Callable, List, Mapping, Optional, Sequence, Tuple, Type, Union
+from typing import Any, Callable, Iterator, List, Mapping, Optional, Sequence, Tuple, Type, Union
 
 import requests
 
@@ -146,6 +146,72 @@ def delete(
     ) -> Any:
         return self.send_request(requests.delete, path, body)
 
+    def post_stream(
+        self,
+        path: str,
+        body: Optional[
+            Union[Mapping[str, Any], Sequence[Mapping[str, Any]], List[str], bytes, str]
+        ] = None,
+        content_type: Optional[str] = "application/json",
+        *,
+        serializer: Optional[Type[json.JSONEncoder]] = None,
+    ) -> requests.Response:
+        """Send a POST request with streaming enabled.
+        
+        Returns the raw response object for streaming consumption.
+        """
+        if content_type:
+            self.headers["Content-Type"] = content_type
+        try:
+            request_path = self.config.url + "/" + path
+            
+            if isinstance(body, bytes):
+                response = requests.post(
+                    request_path,
+                    timeout=self.config.timeout,
+                    headers=self.headers,
+                    data=body,
+                    stream=True,
+                )
+            else:
+                serialize_body = isinstance(body, dict) or body
+                data = (
+                    json.dumps(body, cls=serializer)
+                    if isinstance(body, bool) or serialize_body
+                    else "" if body == "" else "null"
+                )
+
+                response = requests.post(
+                    request_path, 
+                    timeout=self.config.timeout, 
+                    headers=self.headers, 
+                    data=data,
+                    stream=True,
+                )
+            
+            # For streaming responses, we validate status but don't parse JSON
+            if not response.ok:
+                response.raise_for_status()
+            
+            return response
+
+        except requests.exceptions.Timeout as err:
+            raise MeilisearchTimeoutError(str(err)) from err
+        except requests.exceptions.ConnectionError as err:
+            raise MeilisearchCommunicationError(str(err)) from err
+        except requests.exceptions.HTTPError as err:
+            raise MeilisearchApiError(str(err), response) from err
+        except requests.exceptions.InvalidSchema as err:
+            if "://" not in self.config.url:
+                raise MeilisearchCommunicationError(
+                    f"""
+                    Invalid URL {self.config.url}, no scheme/protocol supplied.
+                    Did you mean https://{self.config.url}?
+                    """
+                ) from err
+
+            raise MeilisearchCommunicationError(str(err)) from err
+
     @staticmethod
     def __to_json(request: requests.Response) -> Any:
         if request.content == b"":
diff --git a/meilisearch/client.py b/meilisearch/client.py
@@ -8,12 +8,12 @@
 import hmac
 import json
 import re
-from typing import Any, Dict, List, Mapping, MutableMapping, Optional, Sequence, Tuple, Union
+from typing import Any, Dict, Iterator, List, Mapping, MutableMapping, Optional, Sequence, Tuple, Union
 from urllib import parse
 
 from meilisearch._httprequests import HttpRequests
 from meilisearch.config import Config
-from meilisearch.errors import MeilisearchError
+from meilisearch.errors import MeilisearchApiError, MeilisearchCommunicationError, MeilisearchError
 from meilisearch.index import Index
 from meilisearch.models.key import Key, KeysResults
 from meilisearch.models.task import Batch, BatchResults, Task, TaskInfo, TaskResults
@@ -795,6 +795,156 @@ def get_all_networks(self) -> Dict[str, str]:
         """
         return self.http.get(path=f"{self.config.paths.network}")
 
+    def create_chat_completion(
+        self,
+        workspace_uid: str,
+        messages: List[Dict[str, str]],
+        model: str = "gpt-3.5-turbo",
+        stream: bool = True,
+    ) -> Iterator[Dict[str, Any]]:
+        """Streams a chat completion from the Meilisearch chat API.
+
+        Parameters
+        ----------
+        workspace_uid:
+            Unique identifier of the chat workspace to use.
+        messages:
+            List of message dicts (e.g. {"role": "user", "content": "..."}) comprising the chat history.
+        model:
+            The model name to use for completion (should correspond to the LLM in workspace settings).
+        stream:
+            Whether to stream the response. Must be True for now (only streaming is supported).
+
+        Returns
+        -------
+        chunks:
+            Parsed chunks of the completion as Python dicts. Each chunk is a partial response (in OpenAI format).
+            Iteration ends when the completion is done.
+
+        Raises
+        ------
+        MeilisearchApiError
+            An error containing details about why Meilisearch can't process your request. Meilisearch error codes are described here: https://www.meilisearch.com/docs/reference/errors/error_codes#meilisearch-errors
+        MeilisearchCommunicationError
+            If a network error occurs.
+        ValueError
+            If stream=False is passed (not currently supported).
+        """
+        if not stream:
+            # The API currently only supports streaming responses:
+            raise ValueError("Non-streaming chat completions are not supported. Use stream=True.")
+
+        payload = {
+            "model": model,
+            "messages": messages,
+            "stream": True
+        }
+        
+        # Construct the URL for the chat completions route.
+        endpoint = f"chats/{workspace_uid}/chat/completions"
+        
+        # Initiate the HTTP POST request in streaming mode.
+        response = self.http.post_stream(endpoint, body=payload)
+        
+        try:
+            # Iterate over the streaming response lines
+            for raw_line in response.iter_lines():
+                if raw_line is None or raw_line == b'':
+                    continue  
+                
+                line = raw_line.decode('utf-8')
+                if line.startswith("data: "):
+                    data = line[len("data: "):]  
+                    if data.strip() == "[DONE]":
+                        break 
+                    
+                    try:
+                        chunk = json.loads(data)
+                        yield chunk
+                    except json.JSONDecodeError as e:
+                    
+                        raise MeilisearchCommunicationError(f"Failed to parse chat chunk: {e}")
+        finally:
+            response.close()
+
+    def get_chat_workspaces(
+        self,
+        *,
+        offset: Optional[int] = None,
+        limit: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        """Get all chat workspaces.
+
+        Parameters
+        ----------
+        offset (optional):
+            Number of workspaces to skip.
+        limit (optional):
+            Maximum number of workspaces to return.
+
+        Returns
+        -------
+        workspaces
+            Dictionary containing the list of chat workspaces and pagination information.
+
+        Raises
+        ------
+        MeilisearchApiError
+            An error containing details about why Meilisearch can't process your request. Meilisearch error codes are described here: https://www.meilisearch.com/docs/reference/errors/error_codes#meilisearch-errors
+        """
+        q = []
+        if offset is not None:
+            q.append(f"offset={offset}")
+        if limit is not None:
+            q.append(f"limit={limit}")
+        path = "chats" + ("?" + "&".join(q) if q else "")
+        return self.http.get(path)
+
+    def get_chat_workspace_settings(self, workspace_uid: str) -> Dict[str, Any]:
+        """Get the settings for a specific chat workspace.
+
+        Parameters
+        ----------
+        workspace_uid:
+            Unique identifier of the chat workspace.
+
+        Returns
+        -------
+        settings:
+            Dictionary containing the workspace settings.
+
+        Raises
+        ------
+        MeilisearchApiError
+            An error containing details about why Meilisearch can't process your request. Meilisearch error codes are described here: https://www.meilisearch.com/docs/reference/errors/error_codes#meilisearch-errors
+        """
+        return self.http.get(f"chats/{workspace_uid}/settings")
+
+
+    def update_chat_workspace_settings(
+        self, workspace_uid: str, settings: Mapping[str, Any]
+    ) -> Dict[str, Any]:
+        """Update the settings for a specific chat workspace.
+
+        Parameters
+        ----------
+        workspace_uid:
+            Unique identifier of the chat workspace.
+        settings:
+            Dictionary containing the settings to update.
+
+        Returns
+        -------
+        settings:
+            Dictionary containing the updated workspace settings.
+
+        Raises
+        ------
+        MeilisearchApiError
+            An error containing details about why Meilisearch can't process your request. Meilisearch error codes are described here: https://www.meilisearch.com/docs/reference/errors/error_codes#meilisearch-errors
+        """
+        return self.http.patch(f"chats/{workspace_uid}/settings", body=settings)
+
     @staticmethod
     def _base64url_encode(data: bytes) -> str:
         return base64.urlsafe_b64encode(data).decode("utf-8").replace("=", "")
diff --git a/tests/client/test_chat_completions.py b/tests/client/test_chat_completions.py
@@ -0,0 +1,144 @@
+# pylint: disable=invalid-name
+
+import json
+from unittest.mock import patch
+
+import pytest
+import requests
+
+import meilisearch
+from meilisearch.errors import MeilisearchApiError, MeilisearchCommunicationError
+from tests import BASE_URL, MASTER_KEY
+
+
+class MockStreamingResponse:
+    """Mock response object for testing streaming functionality."""
+    
+    def __init__(self, lines, ok=True, status_code=200, text=""):
+        self.lines = lines
+        self.ok = ok
+        self.status_code = status_code
+        self.text = text
+        self._closed = False
+    
+    def iter_lines(self):
+        """Simulate iter_lines() method of requests.Response."""
+        for line in self.lines:
+            yield line
+    
+    def close(self):
+        """Simulate close() method of requests.Response."""
+        self._closed = True
+    
+    def raise_for_status(self):
+        """Simulate raise_for_status() method of requests.Response."""
+        if not self.ok:
+            raise requests.exceptions.HTTPError(f"HTTP {self.status_code}")
+
+
+def test_create_chat_completion_basic_stream(client):
+    """Test basic streaming functionality with successful response."""
+    dummy_lines = [
+        b'data: {"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"delta":{"content":"Hello"}}]}',
+        b'data: {"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"delta":{"content":" world"}}]}',
+        b'data: [DONE]'
+    ]
+    mock_resp = MockStreamingResponse(dummy_lines)
+    
+    with patch.object(client.http, 'post_stream', return_value=mock_resp) as mock_post:
+        messages = [{"role": "user", "content": "Hi"}]
+        chunks = list(client.create_chat_completion("my-assistant", messages=messages))
+        
+        # Verify the HTTP call was made correctly
+        mock_post.assert_called_once_with(
+            "chats/my-assistant/chat/completions",
+            body={
+                "model": "gpt-3.5-turbo",
+                "messages": messages,
+                "stream": True
+            }
+        )
+        
+        # Verify the chunks are parsed correctly
+        assert len(chunks) == 2
+        assert chunks[0]["choices"][0]["delta"]["content"] == "Hello"
+        assert chunks[1]["choices"][0]["delta"]["content"] == " world"
+        assert mock_resp._closed
+
+
+def test_create_chat_completion_stream_false_raises_error(client):
+    """Test that stream=False raises ValueError."""
+    messages = [{"role": "user", "content": "Test"}]
+    
+    with pytest.raises(ValueError, match="Non-streaming chat completions are not supported"):
+        list(client.create_chat_completion("my-assistant", messages=messages, stream=False))
+
+
+def test_create_chat_completion_json_decode_error(client):
+    """Test that malformed JSON raises MeilisearchCommunicationError."""
+    dummy_lines = [
+        b'data: {"invalid": json}',  # Malformed JSON
+    ]
+    mock_resp = MockStreamingResponse(dummy_lines)
+    
+    with patch.object(client.http, 'post_stream', return_value=mock_resp):
+        messages = [{"role": "user", "content": "Test"}]
+        
+        with pytest.raises(MeilisearchCommunicationError, match="Failed to parse chat chunk"):
+            list(client.create_chat_completion("my-assistant", messages=messages))
+
+
+def test_create_chat_completion_http_error_propagated(client):
+    """Test that HTTP errors from post_stream are properly propagated."""
+    with patch.object(client.http, 'post_stream') as mock_post:
+        error_response = MockStreamingResponse([], ok=False, status_code=400, text='{"message": "API Error"}')
+        mock_post.side_effect = MeilisearchApiError("API Error", error_response)
+        messages = [{"role": "user", "content": "Test"}]
+        
+        with pytest.raises(MeilisearchApiError, match="API Error"):
+            list(client.create_chat_completion("my-assistant", messages=messages))
+
+
+def test_get_chat_workspaces(client):
+    """Test basic get_chat_workspaces functionality."""
+    mock_response = {
+        "results": [
+            {"uid": "workspace1", "name": "My Workspace", "model": "gpt-3.5-turbo"},
+            {"uid": "workspace2", "name": "Another Workspace", "model": "gpt-4"}
+        ],
+        "offset": 0,
+        "limit": 20,
+        "total": 2
+    }
+    
+    with patch.object(client.http, 'get', return_value=mock_response) as mock_get:
+        result = client.get_chat_workspaces()
+        
+        # Verify the HTTP call was made correctly
+        mock_get.assert_called_once_with("chats")
+        
+        # Verify the response is returned as-is
+        assert result == mock_response
+
+
+def test_update_chat_workspace_settings(client):
+    """Test basic update_chat_workspace_settings functionality."""
+    mock_response = {
+        "model": "gpt-4-turbo",
+        "temperature": 0.8,
+        "max_tokens": 1500
+    }
+    
+    settings_update = {
+        "temperature": 0.8,
+        "max_tokens": 1500
+    }
+    
+    with patch.object(client.http, 'patch', return_value=mock_response) as mock_patch:
+        result = client.update_chat_workspace_settings("my-workspace", settings_update)
+        
+        # Verify the HTTP call was made correctly
+        mock_patch.assert_called_once_with("chats/my-workspace/settings", body=settings_update)
+        
+        # Verify the response is returned as-is
+        assert result == mock_response
