Added Lemmy comment streaming, further testing needed

Added small api wrapper to stream lemmy comments like Praw comments. The wrapper still needs to be tested.

.gitignore

@@ -1,6 +1,7 @@
 passwords.py
+async_lemmy_test.py
 creds.json
-__pycache__/*
+__pycache__
 venv
 *.json
 *.ipynb

.python-version

@@ -0,0 +1 @@
+3.11

async_lemmy_py/__init__.py

@@ -0,0 +1 @@
+from async_lemmy_py.async_lemmy import AsyncLemmyPy

async_lemmy_py/async_lemmy.py

@@ -0,0 +1,106 @@
+import asyncio
+import random
+from typing import Any, AsyncIterator, Self
+
+from cachetools import Cache
+
+from async_lemmy_py.models.comment import Comment
+from async_lemmy_py.request_builder import RequestBuilder
+
+
+class AsyncLemmyPy:
+    """A client for interacting with the Lemmy API asynchronously.
+
+    This class provides an asynchronous interface to interact with the Lemmy API, allowing users to stream comments from a specified community. It supports
+    asynchronous context management and can be used within an 'async with' block.
+
+    :param str base_url: The base URL of the Lemmy instance.
+    :param str username: The username for authentication.
+    :param str password: The password for authentication.
+    :ivar RequestBuilder request_builder: An instance of RequestBuilder for building API requests.
+
+    """
+
+    def __init__(self, base_url: str, username: str, password: str) -> None:
+        """Initialize the AsyncLemmyPy instance.
+
+        :param str base_url: The base URL of the Lemmy instance.
+        :param str username: The username for authentication.
+        :param str password: The password for authentication.
+
+        """
+
+        self.request_builder = RequestBuilder(base_url, username, password)
+
+    async def __aenter__(self) -> Self:
+        """Enter the asynchronous context.
+
+        :returns: The current instance of AsyncLemmyPy.
+        :rtype: AsyncLemmyPy
+
+        """
+        return self
+
+    async def __aexit__(self, *_: Any) -> None:
+        """Exit the asynchronous context and close the request builder."""
+        await self.request_builder.close()
+
+    async def stream_comments(self) -> AsyncIterator[Comment]:
+        """Asynchronously stream comments from a Lemmy community.
+
+        :yields: A Comment object representing a comment from the Lemmy community.
+        :rtype: Comment
+
+        """
+        exponential_counter = ExponentialCounter(max_counter=16)
+        seen_comments: Cache[int, Comment] = Cache(maxsize=600)
+        skip_first = True
+        found = False
+
+        while True:
+            comments = await self.request_builder.get(
+                "comment/list", params={"type_": "Local", "sort": "New", "max_depth": 8, "page": 1, "community_id": 2, "community_name": "pcm"}
+            )
+
+            for raw_comment in comments.get("comments", []):
+                comment = Comment.from_dict(raw_comment)
+                if comment.id in seen_comments:
+                    continue
+                found = True
+                seen_comments.update({comment.id: comment})
+                if not skip_first:
+                    yield comment
+
+            skip_first = False
+            if found:
+                exponential_counter.reset()
+            else:
+                await asyncio.sleep(exponential_counter.counter())
+
+
+class ExponentialCounter:
+    """A class to provide an exponential counter with jitter."""
+
+    def __init__(self, max_counter: int):
+        """Initialize an :class:`.ExponentialCounter` instance.
+
+        :param max_counter: The maximum base value.
+
+            .. note::
+
+                The computed value may be 3.125% higher due to jitter.
+
+        """
+        self._base = 1
+        self._max = max_counter
+
+    def counter(self) -> int | float:
+        """Increment the counter and return the current value with jitter."""
+        max_jitter = self._base / 16.0
+        value = self._base + random.random() * max_jitter - max_jitter / 2  # noqa: S311
+        self._base = min(self._base * 2, self._max)
+        return value
+
+    def reset(self) -> None:
+        """Reset the counter to 1."""
+        self._base = 1

async_lemmy_py/models/comment.py

@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from datetime import datetime
+
+from async_lemmy_py.models.community import Community
+from async_lemmy_py.models.post import Post
+from async_lemmy_py.models.user import User
+from typing import Self, cast
+
+
+@dataclass
+class Comment:
+    id: int
+    creator_id: int
+    post_id: int
+    content: str
+    removed: bool
+    published: datetime
+    deleted: bool
+    ap_id: str
+    local: bool
+    path: str
+    distinguished: bool
+    language_id: int
+    input_dict: dict[str, str | int | bool]
+
+    post: Post
+    community: Community
+    user: User
+
+    @classmethod
+    def from_dict(cls, data: dict[str, dict[str, str | int | bool]]) -> Self:
+        comment_data = data["comment"]
+        published_datetime = datetime.fromisoformat(cast(str, comment_data.get("published", "1969-12-31T19:00:00")))
+        return cls(
+            id=cast(int, comment_data.get("id", 0)),
+            creator_id=cast(int, comment_data.get("creator_id", 0)),
+            post_id=cast(int, comment_data.get("post_id", 0)),
+            content=cast(str, comment_data.get("content", "")),
+            removed=cast(bool, comment_data.get("removed", False)),
+            published=published_datetime,
+            deleted=cast(bool, comment_data.get("deleted", False)),
+            ap_id=cast(str, comment_data.get("ap_id", "")),
+            local=cast(bool, comment_data.get("local", False)),
+            path=cast(str, comment_data.get("path", "")),
+            distinguished=cast(bool, comment_data.get("distinguished", False)),
+            language_id=cast(int, comment_data.get("language_id", 0)),
+            input_dict=comment_data,
+            post=Post.from_dict(data["post"]),
+            community=Community.from_dict(data["community"]),
+            user=User.from_dict(data["creator"]),
+        )

async_lemmy_py/models/community.py

@@ -0,0 +1,38 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Self
+
+
+@dataclass
+class Community:
+    id: int
+    name: str
+    title: str
+    description: str
+    removed: bool
+    published: datetime
+    updated: datetime
+    deleted: bool
+    nsfw: bool
+    actor_id: str
+    local: bool
+    icon: str
+    banner: str
+    hidden: bool
+    posting_restricted_to_mods: bool
+    instance_id: int
+
+    @classmethod
+    def from_dict(cls, data: dict[Any, Any]) -> Self:
+        data_copy = data.copy()
+
+        published_str = data_copy.get("published")
+        updated_str = data_copy.get("updated")
+
+        if published_str:
+            data_copy["published"] = datetime.fromisoformat(published_str)
+
+        if updated_str:
+            data_copy["updated"] = datetime.fromisoformat(updated_str)
+
+        return cls(**data_copy)

async_lemmy_py/models/post.py

@@ -0,0 +1,39 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional, Self, Any
+
+
+@dataclass
+class Post:
+    id: int
+    name: str
+    creator_id: int
+    community_id: int
+    removed: bool
+    locked: bool
+    published: datetime
+    deleted: bool
+    nsfw: bool
+    ap_id: str
+    local: bool
+    language_id: int
+    featured_community: bool
+    featured_local: bool
+    body: Optional[str] = None
+    embed_description: Optional[str] = None
+    embed_title: Optional[str] = None
+    embed_video_url: Optional[str] = None
+    thumbnail_url: Optional[str] = None
+    updated: Optional[str] = None
+    url: Optional[str] = None
+
+    @classmethod
+    def from_dict(cls, data: dict[Any, Any]) -> Self:
+        data_copy = data.copy()
+
+        published_str = data_copy.get("published")
+
+        if published_str:
+            data_copy["published"] = datetime.fromisoformat(published_str)
+
+        return cls(**data_copy)

async_lemmy_py/models/user.py

@@ -0,0 +1,36 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional, Self, Any
+
+
+@dataclass
+class User:
+    id: int
+    name: str
+    banned: bool
+    published: datetime
+    actor_id: str
+    local: bool
+    deleted: bool
+    admin: bool
+    bot_account: bool
+    instance_id: int
+    bio: Optional[str] = None
+    inbox_url: Optional[str] = None
+    matrix_user_id: Optional[str] = None
+    display_name: Optional[str] = None
+    avatar: Optional[str] = None
+    ban_expires: Optional[str] = None
+    banner: Optional[str] = None
+    updated: Optional[str] = None
+
+    @classmethod
+    def from_dict(cls, data: dict[Any, Any]) -> Self:
+        data_copy = data.copy()
+
+        published_str = data_copy.get("published")
+
+        if published_str:
+            data_copy["published"] = datetime.fromisoformat(published_str)
+
+        return cls(**data_copy)

async_lemmy_py/request_builder.py

@@ -0,0 +1,104 @@
+from typing import Optional, Any
+
+from aiohttp import ClientResponse, ClientResponseError, ClientSession
+
+
+class RequestBuilder:
+    def __init__(self, base_url: str, username: str, password: str) -> None:
+        """Initialize the RequestBuilder.
+
+        :param base_url: The base URL for API requests.
+        :param username: The username for authentication.
+        :param password: The password for authentication.
+
+        """
+        self.base_url: str = base_url
+        self.username: str = username
+        self.password: str = password
+        self.jwt_token: Optional[str] = None
+
+        # Initialize aiohttp ClientSession with default headers
+        self.client_session: ClientSession = ClientSession(headers={"accept": "application/json", "content-type": "application/json"})
+
+    async def get_jwt_token(self) -> None:
+        """Get JWT token by sending a POST request to the login endpoint.
+
+        The token will be stored in the instance variable `jwt_token`.
+
+        """
+        auth = {"password": self.password, "username_or_email": self.username}
+        async with self.client_session.post(f"{self.base_url}/api/v3/user/login", json=auth) as resp:
+            data = await resp.json()
+            self.jwt_token = data.get("jwt")
+
+    async def close(self) -> None:
+        """Close the aiohttp ClientSession."""
+        await self.client_session.close()
+
+    async def get(self, endpoint: str, params: Optional[dict[Any, Any]] = None) -> dict[Any, Any]:
+        """Perform an HTTP GET request.
+
+        If jwt_token is None, it will first call get_jwt_token to obtain the token.
+
+        :param endpoint: The API endpoint to send the GET request to.
+        :param params: Optional query parameters.
+
+        :returns: JSON response from the server.
+
+        :raises: If the HTTP response status code indicates an error (not in the 2xx range).
+
+        """
+        if self.jwt_token is None:
+            await self.get_jwt_token()
+
+        url: str = f"{self.base_url}/api/v3/{endpoint}"
+        headers = {"Authorization": f"Bearer {self.jwt_token}"}
+        params_with_auth = {"auth": self.jwt_token} if params is None else {**params, "auth": self.jwt_token}
+
+        async with self.client_session.get(url, headers=headers, params=params_with_auth) as resp:
+            return await self._handle_response(resp)
+
+    async def post(
+        self, endpoint: str, params: Optional[dict[Any, Any]] = None, data: Optional[dict[Any, Any]] = None, json: Optional[dict[Any, Any]] = None
+    ) -> dict[Any, Any]:
+        """Perform an HTTP POST request.
+
+        :param endpoint: The API endpoint to send the POST request to.
+        :param params: Optional query parameters.
+        :param data: Optional data for the request body (used for form data).
+        :param json: Optional JSON data for the request body.
+
+        :returns: JSON response from the server.
+
+        :raises: If the HTTP response status code indicates an error (not in the 2xx range).
+
+        """
+        url: str = f"{self.base_url}/api/v3/{endpoint}"
+        headers = {"Authorization": f"Bearer {self.jwt_token}"}
+        params_with_auth = {"auth": self.jwt_token} if params is None else {**params, "auth": self.jwt_token}
+
+        async with self.client_session.post(url, headers=headers, params=params_with_auth, data=data, json=json) as resp:
+            return await self._handle_response(resp)
+
+    async def _handle_response(self, resp: ClientResponse) -> dict[Any, Any]:
+        """Handle the response from the server.
+
+        :param resp: The aiohttp ClientResponse object.
+
+        :returns: JSON response from the server.
+
+        :raises: If the HTTP response status code indicates an error (not in the 2xx range).
+
+        """
+        data: dict[Any, Any] = await resp.json()
+
+        if resp.status >= 300:
+            raise ClientResponseError(
+                request_info=resp.request_info,
+                history=resp.history,
+                status=resp.status,
+                message=f"Request failed with status {resp.status}: {data}",
+                headers=resp.headers,
+            )
+
+        return data

pyproject.toml

@@ -22,3 +22,4 @@ line-length = 160
 [tool.ruff.per-file-ignores]
 "backup_and_cheating/back_up_and_cheating.py" = ["E402"]
 "backup_and_cheating/backup_drive.py" = ["E402"]
+"async_lemmy_py/__init__.py" = ["F401"]