/
more.py
80 lines (69 loc) · 3.28 KB
/
more.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
"""Provide the MoreComments class."""
from typing import TYPE_CHECKING, Any, Dict, List, Union
from ...const import API_PATH
from ..base import AsyncPRAWBase
if TYPE_CHECKING: # pragma: no cover
import asyncpraw
class MoreComments(AsyncPRAWBase):
"""A class indicating there are more comments."""
def __init__(self, reddit: "asyncpraw.Reddit", _data: Dict[str, Any]):
"""Construct an instance of the MoreComments object."""
self.count = self.parent_id = None
self.children = []
super().__init__(reddit, _data=_data)
self._comments = None
self.submission = None
def __eq__(self, other: Union[str, "MoreComments"]) -> bool:
"""Return True if these MoreComments instances are the same."""
if isinstance(other, self.__class__):
return self.count == other.count and self.children == other.children
return super().__eq__(other)
def __lt__(self, other: "MoreComments") -> bool:
"""Provide a sort order on the MoreComments object."""
# To work with heapq a "smaller" item is the one with the most comments. We are
# intentionally making the biggest element the smallest element to turn the
# min-heap implementation in heapq into a max-heap.
return self.count > other.count
def __repr__(self) -> str:
"""Return a string representation of the MoreComments instance."""
children = self.children[:4]
if len(self.children) > 4:
children[-1] = "..."
return f"<{self.__class__.__name__} count={self.count}, children={children!r}>"
async def _continue_comments(self, update):
assert not self.children, "Please file a bug report with Async PRAW."
parent = await self._load_comment(self.parent_id.split("_", 1)[1])
self._comments = parent.replies
if update:
for comment in self._comments:
comment.submission = self.submission
return self._comments
async def _load_comment(self, comment_id):
path = f"{API_PATH['submission'].format(id=self.submission.id)}_/{comment_id}"
_, comments = await self._reddit.get(
path,
params={
"limit": self.submission.comment_limit,
"sort": self.submission.comment_sort,
},
)
assert len(comments.children) == 1, "Please file a bug report with Async PRAW."
return comments.children[0]
async def comments(self, update: bool = True) -> List["asyncpraw.models.Comment"]:
"""Fetch and return the comments for a single MoreComments object."""
if self._comments is None:
if self.count == 0: # Handle "continue this thread"
return await self._continue_comments(update)
assert self.children, "Please file a bug report with Async PRAW."
data = {
"children": ",".join(self.children),
"link_id": self.submission.fullname,
"sort": self.submission.comment_sort,
}
self._comments = await self._reddit.post(
API_PATH["morechildren"], data=data
)
if update:
for comment in self._comments:
comment.submission = self.submission
return self._comments