Comments

piped_api/__init__.py

@@ -0,0 +1,12 @@
+import typing as t
+
+
+from .client import PipedClient
+
+from .models.comments import Comments
+
+
+
+# Supress unused-import warnings:
+if t.TYPE_CHECKING:
+    _ = [PipedClient, Comments]

piped_api/client.py

@@ -0,0 +1,64 @@
+import typing as t
+
+from requests import Session
+
+from .models import BasePipedModel
+from .models.comments import Comments
+
+
+_MDL = t.TypeVar('_MDL', bound=BasePipedModel)
+
+
+
+class PipedClient:
+    """
+        An API client for [Piped](https://piped.kavin.rocks).
+    """
+
+    def __init__(self, base_api_url: str='https://pipedapi.kavin.rocks', session: t.Type[Session]=Session()) -> None:
+        """
+            ### Parameters:
+            - `base_api_url` - The base URL to the instance's API. Trailing slashes will be stripped.
+            - `session` - A class/subclass of `requests.Session` to use for all requests.
+                For example, you could use [requests-cache](https://pypi.org/project/requests-cache/) to make all requests cacheable.
+        """
+
+        self.base_api_url = base_api_url.strip("/")
+        self.session =  session
+
+
+
+    def _get_json(self, uri: str, as_model: t.Optional[_MDL]=None, **kwargs) -> t.Union[_MDL, t.Dict[str, t.Any]]:
+        """
+            Obtains JSON data from specific URI of the Piped API.
+
+            ### Parameters:
+            - `uri` - The URI to get JSON data from
+            - `as_model` - The `BasePipedModel` to load the JSON data into. If this is `None`, the JSON data is returned as a `dict`.
+            - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
+        """
+
+        json = self.session.get(f"{self.base_api_url}{uri}", **kwargs).json()
+
+        if as_model is not None:
+            return as_model(json)
+
+        return json
+
+
+
+    def get_comments(self, video_id: str, nextpage: t.Optional[t.Dict[str, t.Optional[str]]]=None) -> Comments:
+        """
+            Gets a list of comments for a specific video.
+
+            ### Parameters:
+            - `video_id` - The ID of the video to get comments for
+            - `nextpage` - Nextpage data, obtained from `.models.comments.Comments.nextpage` property. If this is `None`, the first page of comments is returned.
+                There are often 20 comments per page.
+        """
+
+        if nextpage is not None:
+            return self._get_json(f"/nextpage/comments/{video_id}", Comments, params={"nextpage": nextpage})
+
+        else:
+            return self._get_json(f"/comments/{video_id}", Comments)

piped_api/models/__init__.py

@@ -0,0 +1,15 @@
+import typing as t
+
+
+class BasePipedModel:
+    """
+        Base class for all Piped models.
+    """
+
+    def __init__(self, data: t.Dict[str, t.Any]) -> None:
+        """
+            ### Parameters:
+            - `data` - The JSON (`dict`) data to initialize the model with.
+        """
+
+        self.data = data

piped_api/models/comments.py

@@ -0,0 +1,141 @@
+import typing as t
+
+from . import BasePipedModel
+
+
+class Comments(BasePipedModel):
+    class Comment(BasePipedModel):
+        @property
+        def author(self) -> str:
+            """
+                The name of the author of the comment
+            """
+
+            return self.data['author']
+
+
+        @property
+        def comment_id(self) -> str:
+            """
+                The comment ID
+            """
+
+            return self.data['commentId']
+
+
+        @property
+        def comment_text(self) -> str:
+            """
+                The text of the comment
+            """
+
+            return self.data['commentText']
+
+
+        @property
+        def commented_time(self) -> str:
+            """
+                The time the comment was made (format: `'x y ago'`)
+            """
+
+            return self.data['commentedTime']
+
+
+        @property
+        def commentor_url(self) -> str:
+            """
+                The URL of the channel that made the comment
+            """
+
+            return self.data['commentorUrl']
+
+
+        @property
+        def replies_page(self) -> t.Optional[str]:
+            """
+                Same as `Comments.nextpage`, but to load replies.
+
+                `None` means that there are no replies.
+            """
+
+            return self.data['repliesPage']
+
+
+        @property
+        def hearted(self) -> bool:
+            """
+                Whether or not the comment has been hearted
+            """
+
+            return self.data['hearted']
+
+
+        @property
+        def like_count(self) -> int:
+            """
+                The number of likes the comment has
+            """
+
+            return self.data['likeCount']
+
+
+        @property
+        def pinned(self) -> bool:
+            """
+                Whether or not the comment is pinned
+            """
+
+            return self.data['pinned']
+
+
+        @property
+        def thumbnail(self) -> str:
+            """
+                The thumbnail of the commentor's channel
+            """
+
+            return self.data['thumbnail']
+
+
+        @property
+        def verified(self) -> bool:
+            """
+                Whether or not the author of the comment is verified
+            """
+
+            return self.data['verified']
+
+
+
+    def get_comments(self) -> t.List[Comment]:
+        """
+            Obtain a list of comments
+        """
+
+        return [self.Comment(comment_json) for comment_json in self.data['comments']]
+
+
+    def __iter__(self) -> t.Iterator[Comment]:
+        iter(self.get_comments())
+
+
+
+    @property
+    def disabled(self) -> bool:
+        """
+            Whether or not the comments are disabled
+        """
+
+        return self.data['disabled']
+
+
+
+    @property
+    def nextpage(self) -> t.Optional[str]:
+        """
+            A JSON encoded page, which is used for the nextpage endpoint.
+
+            If there is no nextpage data, this returns `None`.
+        """
+
+        return self.data['nextpage']