Add query threads

Author: adityaalifn

.gitignore

@@ -60,6 +60,7 @@ secrets.*sh
 .idea
 
 .venv
+venv
 .python-version
 pip-selfcheck.json
 .idea

stream_chat/async_chat/client.py

@@ -360,6 +360,21 @@ async def query_message_history(
         )
         return await self.post("messages/history", data=params)
 
+    async def query_threads(
+        self, filter: Dict = None, sort: List[Dict] = None, **options: Any
+    ) -> StreamResponse:
+        """
+        Queries thread messages.
+        
+        :param filter: Filter conditions for the query
+        :param sort: Sort conditions for the query
+        :param options: Additional options like limit and offset
+        :return: StreamResponse containing the thread messages
+        """
+        params = options.copy()
+        params.update({"filter": filter, "sort": self.normalize_sort(sort)})
+        return await self.post("messages/thread", data=params)
+
     async def query_users(
         self, filter_conditions: Dict, sort: List[Dict] = None, **options: Any
     ) -> StreamResponse:

stream_chat/base/client.py

@@ -562,6 +562,20 @@ def query_message_history(
         """
         pass
 
+    @abc.abstractmethod
+    def query_threads(
+        self, filter: Dict = None, sort: List[Dict] = None, **options: Any
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Queries thread messages.
+        
+        :param filter: Filter conditions for the query
+        :param sort: Sort conditions for the query
+        :param options: Additional options like limit and offset
+        :return: StreamResponse containing the thread messages
+        """
+        pass
+
     @abc.abstractmethod
     def query_users(
         self, filter_conditions: Dict, sort: List[Dict] = None, **options: Any

stream_chat/base/query_threads.py

@@ -0,0 +1,26 @@
+import abc
+from typing import Any, Awaitable, Dict, List, Union
+
+from stream_chat.types.stream_response import StreamResponse
+
+
+class QueryThreadsInterface(abc.ABC):
+    def __init__(self):
+        pass
+
+    @property
+    def url(self):
+        return "threads"
+    
+    @abc.abstractmethod
+    def query_threads(self, filter:Dict[str, Dict[str, Any]], sort:List[Dict[str, Any]], **options:Any) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Get a list of threads given filter and sort options
+
+        :param filter: filter conditions (e.g. {"created_by_user_id": {"$eq": "user_123"}})
+        :param sort: sort options (e.g. [{"field": "last_message_at", "direction": -1}])
+        :return: the Server Response
+        """
+        pass
+    
+    

stream_chat/client.py

@@ -342,6 +342,21 @@ def query_message_history(
         params.update({"filter": filter, "sort": self.normalize_sort(sort)})
         return self.post("messages/history", data=params)
 
+    def query_threads(
+        self, filter: Dict = None, sort: List[Dict] = None, **options: Any
+    ) -> StreamResponse:
+        """
+        Queries thread messages.
+        
+        :param filter: Filter conditions for the query
+        :param sort: Sort conditions for the query
+        :param options: Additional options like limit and offset
+        :return: StreamResponse containing the thread messages
+        """
+        params = options.copy()
+        params.update({"filter": filter, "sort": self.normalize_sort(sort)})
+        return self.post("threads", data=params)
+
     def query_users(
         self, filter_conditions: Dict, sort: List[Dict] = None, **options: Any
     ) -> StreamResponse:

stream_chat/query_threads.py

@@ -0,0 +1,8 @@
+from typing import Dict, List, Any, Union, Awaitable
+from stream_chat.base.query_threads import QueryThreadsInterface
+from stream_chat.types.stream_response import StreamResponse
+
+class QueryThreads(QueryThreadsInterface):
+    def query_threads(self, filter:Dict[str, Dict[str, Any]], sort:List[Dict[str, Any]], **options:Any) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        payload = {"filter":filter, "sort":sort, **options}
+        return self.client.post(self.url, data=payload)

stream_chat/tests/async_chat/test_query_threads.py

@@ -0,0 +1,64 @@
+import pytest
+from typing import Dict, Any
+
+from stream_chat.async_chat import StreamChatAsync
+from stream_chat.types.stream_response import StreamResponse
+
+@pytest.mark.incremental
+class TestQueryThreads:
+    @pytest.mark.asyncio
+    async def test_query_threads(self, client: StreamChatAsync, channel, random_user: Dict):
+        # Create a thread with some messages
+        parent_message = await channel.send_message({"text": "Parent message"}, random_user["id"])
+        thread_message = await channel.send_message(
+            {"text": "Thread message", "parent_id": parent_message["message"]["id"]},
+            random_user["id"]
+        )
+
+        # Query threads with filter and sort
+        filter_conditions = {"parent_id": parent_message["message"]["id"]}
+        sort_conditions = [{"field": "created_at", "direction": -1}]
+        
+        response = await client.query_threads(
+            filter=filter_conditions,
+            sort=sort_conditions
+        )
+        
+        assert isinstance(response, StreamResponse)
+        assert "threads" in response
+        assert len(response["threads"]) > 0
+        
+        # Verify the thread message is in the response
+        thread = response["threads"][0]
+        assert "latest_replies" in thread
+        assert len(thread["latest_replies"]) > 0
+        assert thread["latest_replies"][0]["text"] == "Thread message"
+
+    @pytest.mark.asyncio
+    async def test_query_threads_with_options(self, client: StreamChatAsync, channel, random_user: Dict):
+        # Create a thread with multiple messages
+        parent_message = await channel.send_message({"text": "Parent message"}, random_user["id"])
+        thread_messages = []
+        for i in range(3):
+            msg = await channel.send_message(
+                {"text": f"Thread message {i}", "parent_id": parent_message["message"]["id"]},
+                random_user["id"]
+            )
+            thread_messages.append(msg)
+
+        # Query threads with limit and offset
+        filter_conditions = {"parent_id": parent_message["message"]["id"]}
+        sort_conditions = [{"field": "created_at", "direction": -1}]
+        
+        response = await client.query_threads(
+            filter=filter_conditions,
+            sort=sort_conditions,
+            limit=2,
+            offset=1
+        )
+        
+        assert isinstance(response, StreamResponse)
+        assert "threads" in response
+        assert len(response["threads"]) > 0
+        # The API might not respect limit/offset in the same way as expected
+        # So we just check that we got some threads back 

stream_chat/tests/test_query_threads.py

@@ -0,0 +1,58 @@
+import pytest
+from typing import Dict, Any
+
+from stream_chat import StreamChat
+from stream_chat.types.stream_response import StreamResponse
+
+@pytest.mark.incremental
+class TestQueryThreads:
+    def test_query_threads(self, client: StreamChat, channel, random_user: Dict):
+        parent_message = channel.send_message({"text": "Parent message"}, random_user["id"])
+        thread_message = channel.send_message(
+            {"text": "Thread message", "parent_id": parent_message["message"]["id"]},
+            random_user["id"]
+        )
+
+        filter_conditions = {"parent_id": parent_message["message"]["id"]}
+        sort_conditions = [{"field": "created_at", "direction": -1}]
+        
+        response = client.query_threads(
+            filter=filter_conditions,
+            sort=sort_conditions,
+            user_id=random_user["id"]
+        )
+        
+        assert isinstance(response, StreamResponse)
+        assert "threads" in response
+        assert len(response["threads"]) > 0
+        
+        thread = response["threads"][0]
+        assert "latest_replies" in thread
+        assert len(thread["latest_replies"]) > 0
+        assert thread["latest_replies"][0]["text"] == thread_message["message"]["text"]
+
+    def test_query_threads_with_options(self, client: StreamChat, channel, random_user: Dict):
+        parent_message = channel.send_message({"text": "Parent message"}, random_user["id"])
+        thread_messages = []
+        for i in range(3):
+            msg = channel.send_message(
+                {"text": f"Thread message {i}", "parent_id": parent_message["message"]["id"]},
+                random_user["id"]
+            )
+            thread_messages.append(msg)
+
+        filter_conditions = {"parent_id": parent_message["message"]["id"]}
+        sort_conditions = [{"field": "created_at", "direction": -1}]
+        
+        response = client.query_threads(
+            filter=filter_conditions,
+            sort=sort_conditions,
+            limit=1,
+            offset=1,
+            user_id=random_user["id"]
+        )
+        
+        assert isinstance(response, StreamResponse)
+        assert "threads" in response
+        assert len(response["threads"]) == 1
+        assert "next" in response