api: Use batch-based narrow updates for marking messages as read.

Fixes #820.

Author: abdallah-9a

zulip/zulip/__init__.py

@@ -10,6 +10,7 @@
 import traceback
 import types
 import urllib.parse
+import warnings
 from configparser import ConfigParser
 from datetime import datetime
 from typing import (
@@ -901,19 +902,58 @@ def delete_message(self, message_id: int) -> Dict[str, Any]:
         """
         return self.call_endpoint(url=f"messages/{message_id}", method="DELETE")
 
+    def update_message_flags_for_narrow(self, update_data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        See https://zulip.com/api/update-message-flags-for-narrow for usage.
+        """
+        return self.call_endpoint(url="messages/flags/narrow", method="POST", request=update_data)
+
     def update_message_flags(self, update_data: Dict[str, Any]) -> Dict[str, Any]:
         """
         See examples/update-flags for example usage.
         """
         return self.call_endpoint(url="messages/flags", method="POST", request=update_data)
 
+    def _mark_as_read_using_narrow(self, narrow: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        Helper method to batch mark messages as read using a narrow.
+        This prevents server-side timeouts when marking a large number of messages
+        as read (fixes Issue #820) by paginating through unread messages.
+        """
+        request: Dict[str, Any] = {
+            "anchor": "first_unread",
+            "num_before": 0,
+            "num_after": 1000,
+            "narrow": narrow,
+            "op": "add",
+            "flag": "read",
+        }
+        while True:
+            res = self.update_message_flags_for_narrow(request)
+            if res.get("result") != "success":
+                return res
+            if res.get("found_newest", True):
+                break
+            if res.get("last_processed_id") is None:
+                break
+            request["anchor"] = res["last_processed_id"]
+        return {"result": "success", "msg": "", "complete": True}
+
     def mark_all_as_read(self) -> Dict[str, Any]:
         """
         Example usage:
 
         >>> client.mark_all_as_read()
         {'result': 'success', 'msg': ''}
         """
+        warnings.warn(
+            "`mark_all_as_read` is deprecated. Use `update_message_flags_for_narrow` instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        if self.feature_level >= 155:
+            return self._mark_as_read_using_narrow([{"operator": "is", "operand": "unread"}])
+
         return self.call_endpoint(
             url="mark_all_as_read",
             method="POST",
@@ -926,6 +966,19 @@ def mark_stream_as_read(self, stream_id: int) -> Dict[str, Any]:
         >>> client.mark_stream_as_read(42)
         {'result': 'success', 'msg': ''}
         """
+        warnings.warn(
+            "`mark_stream_as_read` is deprecated. Use `update_message_flags_for_narrow` instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        if self.feature_level >= 155:
+            return self._mark_as_read_using_narrow(
+                [
+                    {"operator": "stream", "operand": stream_id},
+                    {"operator": "is", "operand": "unread"},
+                ]
+            )
+
         return self.call_endpoint(
             url="mark_stream_as_read",
             method="POST",
@@ -936,9 +989,23 @@ def mark_topic_as_read(self, stream_id: int, topic_name: str) -> Dict[str, Any]:
         """
         Example usage:
 
-        >>> client.mark_all_as_read(42, 'new coffee machine')
+        >>> client.mark_topic_as_read(42, 'new coffee machine')
         {'result': 'success', 'msg': ''}
         """
+        warnings.warn(
+            "`mark_topic_as_read` is deprecated. Use `update_message_flags_for_narrow` instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        if self.feature_level >= 155:
+            return self._mark_as_read_using_narrow(
+                [
+                    {"operator": "stream", "operand": stream_id},
+                    {"operator": "topic", "operand": topic_name},
+                    {"operator": "is", "operand": "unread"},
+                ]
+            )
+
         return self.call_endpoint(
             url="mark_topic_as_read",
             method="POST",