Made a special stream for r/all that does handling of skipped items

CHANGES.rst

@@ -31,6 +31,10 @@ Unreleased
 * The parameter ``text`` for methods
   :meth:`.SubredditLinkFlairTemplates.update` and
   :meth:`.SubredditRedditorFlairTemplates.update` is no longer required.
+* Subreddit streams streaming from r/all can use a new streamer class that will
+  auto-fill items that aren't yielded from Reddit. However, there are
+  problems with the new stream, so usage of the stream requires the parameter
+  ``use_new_stream`` to be set to True (False by default).
 
 **Removed**
 

praw/models/helpers.py

@@ -177,10 +177,14 @@ def create(
 class SubredditHelper(PRAWBase):
     """Provide a set of functions to interact with Subreddits."""
 
-    def __call__(self, display_name: str) -> Subreddit:
+    def __call__(
+        self, display_name: str, use_new_stream: bool = False
+    ) -> Subreddit:
         """Return a lazy instance of :class:`~.Subreddit`.
 
         :param display_name: The name of the subreddit.
+        :param use_new_stream: Use the new subreddit stream. Please read the
+            note on :class:`.Subreddit` before setting this parameter to True.
         """
         lower_name = display_name.lower()
 
@@ -189,7 +193,11 @@ def __call__(self, display_name: str) -> Subreddit:
         if lower_name == "randnsfw":
             return self._reddit.random_subreddit(nsfw=True)
 
-        return Subreddit(self._reddit, display_name=display_name)
+        return Subreddit(
+            self._reddit,
+            display_name=display_name,
+            use_new_stream=use_new_stream,
+        )
 
     def create(
         self,

praw/models/reddit/subreddit.py

@@ -20,7 +20,7 @@
 from ...util.cache import cachedproperty
 from ..listing.generator import ListingGenerator
 from ..listing.mixins import SubredditListingMixin
-from ..util import permissions_string, stream_generator
+from ..util import permissions_string, stream_generator, r_all_streamer
 from .base import RedditBase
 from .emoji import SubredditEmoji
 from .mixins import FullnameMixin, MessageableMixin
@@ -481,11 +481,21 @@ def wiki(self):
         """
         return SubredditWiki(self)
 
-    def __init__(self, reddit, display_name=None, _data=None):
+    def __init__(
+        self, reddit, display_name=None, _data=None, use_new_stream=False
+    ):
         """Initialize a Subreddit instance.
 
         :param reddit: An instance of :class:`~.Reddit`.
         :param display_name: The name of the subreddit.
+        :param use_new_stream: Use the new SubredditStreams.
+
+        .. note:: The new streams will provide lazy comments that were not sent
+            by Reddit's API, and therefore will not be pre-filled like normal
+            comments. This will lead to extra API requests, and also include
+            comments that have been automatically removed by AutoModerator.
+            Any bot built with the new stream will have to handle this.
+            Therefore, the streams will only be used when explicitly asked for.
 
         .. note:: This class should not be initialized directly. Instead obtain
            an instance via: ``reddit.subreddit('subreddit_name')``
@@ -498,6 +508,7 @@ def __init__(self, reddit, display_name=None, _data=None):
         super().__init__(reddit, _data=_data)
         if display_name:
             self.display_name = display_name
+        self._use_new_stream = use_new_stream
         self._path = API_PATH["subreddit"].format(subreddit=self)
 
     def _fetch_info(self):
@@ -2850,7 +2861,14 @@ def comments(self, **stream_options):
                print(comment)
 
         """
-        return stream_generator(self.subreddit.comments, **stream_options)
+        return (
+            stream_generator(self.subreddit.comments, **stream_options)
+            if (
+                self.subreddit.__dict__.get("display_name", "None") != "all"
+                or not self.subreddit.__dict__.get("_use_new_stream", False)
+            )
+            else r_all_streamer(self.subreddit.comments, **stream_options)
+        )
 
     def submissions(self, **stream_options):
         """Yield new submissions as they become available.
@@ -2872,7 +2890,14 @@ def submissions(self, **stream_options):
                print(submission)
 
         """
-        return stream_generator(self.subreddit.new, **stream_options)
+        return (
+            stream_generator(self.subreddit.new, **stream_options)
+            if (
+                self.subreddit.__dict__.get("display_name", "None") != "all"
+                or not self.subreddit.__dict__.get("_use_new_stream", False)
+            )
+            else r_all_streamer(self.subreddit.new, **stream_options)
+        )
 
 
 class SubredditStylesheet:

praw/models/util.py

@@ -3,6 +3,8 @@
 import time
 from typing import Any, Callable, Generator, List, Optional, Set
 
+from ..util import int_to_base36
+
 
 class BoundedSet:
     """A set with a maximum size that evicts the oldest items when necessary.
@@ -207,3 +209,38 @@ def stream_generator(
                 yield None
             else:
                 time.sleep(exponential_counter.counter())
+
+
+def r_all_streamer(
+    function: Callable[[Any], Any],
+    pause_after: Optional[int] = None,
+    skip_existing: bool = False,
+    attribute_name: str = "fullname",
+    exclude_before: bool = False,
+    **function_kwargs: Any
+) -> Generator[Any, None, None]:
+    """Stream class that handles r/all streams.
+
+    Please refer to the documentation for :func:`.stream_generator`.
+    """
+    prev = 0
+    stream = stream_generator(
+        function,
+        pause_after=pause_after,
+        skip_existing=skip_existing,
+        attribute_name=attribute_name,
+        exclude_before=exclude_before,
+        **function_kwargs
+    )
+    for item in stream:
+        if hasattr(item, "id"):
+            if prev == 0:
+                prev = int(item.id, base=36)
+            cur = int(item.id, base=36)
+            if cur - prev > 1:
+                for idnum in range(prev + 1, cur):
+                    reddit = item._reddit
+                    itemtype = item.__class__
+                    yield itemtype(reddit, id=int_to_base36(idnum))
+            prev = cur
+        yield item

praw/util/__init__.py

@@ -2,3 +2,4 @@
 
 from .cache import cachedproperty  # noqa: F401
 from .snake import camel_to_snake, snake_case_keys  # noqa: F401
+from .base36 import int_to_base36  # noqa: F401

praw/util/base36.py

@@ -0,0 +1,17 @@
+"""Provides a function to convert from int to base36."""
+
+
+def int_to_base36(num):
+    """Convert a positive integer into a base36 string.
+
+    :param num: The number to convert
+    :returns: A base36 string
+    """
+    assert num >= 0
+    digits = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+
+    res = ""
+    while not res or num > 0:
+        num, i = divmod(num, 36)
+        res = digits[i] + res
+    return res.lower()

tests/integration/models/reddit/test_subreddit.py

@@ -1777,6 +1777,18 @@ def test_comments__with_skip_existing(self, _):
             # that there are at least 400 comments in the stream.
             assert count < 400
 
+    @mock.patch("time.sleep", return_value=None)
+    def test_comments_all(self, _):
+        with self.recorder.use_cassette("TestSubredditStreams.comments_new"):
+            generator = self.reddit.subreddit(
+                "all", use_new_stream=True
+            ).stream.comments()
+            items = [generator.__next__() for i in range(400)]
+            for i in range(400 - 1):
+                cur = items[i]
+                next = items[i + 1]
+                assert (int(next.id, base=36) - int(cur.id, base=36)) <= 1
+
     def test_submissions(self):
         with self.recorder.use_cassette("TestSubredditStreams.submissions"):
             generator = self.reddit.subreddit("all").stream.submissions()
@@ -1811,6 +1823,20 @@ def test_submissions__with_pause_and_skip_after(self, _):
                 submission = next(generator)
             assert submission_count < 100
 
+    @mock.patch("time.sleep", return_value=None)
+    def test_submissions_all(self, _):
+        with self.recorder.use_cassette(
+            "TestSubredditStreams.submissions_new"
+        ):
+            generator = self.reddit.subreddit(
+                "all", use_new_stream=True
+            ).stream.submissions()
+            items = [generator.__next__() for i in range(101)]
+            for i in range(101 - 1):
+                cur = items[i]
+                next = items[i + 1]
+                assert (int(next.id, base=36) - int(cur.id, base=36)) <= 1
+
 
 class TestSubredditModerationStreams(IntegrationTest):
     @property

tests/unit/models/test_util.py

@@ -1,9 +1,20 @@
 """Test praw.models.util."""
-from praw.models.util import ExponentialCounter, permissions_string
+from praw.models.util import BoundedSet, ExponentialCounter, permissions_string
 
 from .. import UnitTest
 
 
+class TestBoundedSet(UnitTest):
+    def test_limit(self):
+        bset = BoundedSet(2)
+        for i in range(10):
+            bset.add(i)
+        assert 8 in bset
+        assert 9 in bset
+        assert 7 not in bset
+        assert 0 not in bset
+
+
 class TestExponentialCounter(UnitTest):
     MAX_DELTA = 1.0 / 32