From 246cdf51fa94ebc2b08bd9d79e2c1f5af5ff27be Mon Sep 17 00:00:00 2001 From: retoor Date: Sat, 2 Aug 2025 08:03:19 +0200 Subject: [PATCH] Update client. --- src/devranta/api.py | 445 ++++++++++++++++++++++++++++++++++++++------ 1 file changed, 389 insertions(+), 56 deletions(-) diff --git a/src/devranta/api.py b/src/devranta/api.py index a003bcf..a3cdf63 100644 --- a/src/devranta/api.py +++ b/src/devranta/api.py @@ -1,28 +1,124 @@ -from typing import Literal, Optional +from __future__ import annotations +from typing import Any, Dict, List, Literal, Optional, TypedDict, Union + import aiohttp from enum import Enum + class VoteReason(Enum): + """Enumeration for reasons when down-voting a rant or comment.""" NOT_FOR_ME = 0 REPOST = 1 OFFENSIVE_SPAM = 2 +# --- TypedDicts for API Responses --- + +class AuthToken(TypedDict): + id: int + key: str + expire_time: int + user_id: int + +class LoginResponse(TypedDict): + success: bool + auth_token: AuthToken + +class Image(TypedDict): + url: str + width: int + height: int + +class UserAvatar(TypedDict): + b: str # background color + i: NotRequired[str] # image identifier + +class Rant(TypedDict): + id: int + text: str + score: int + created_time: int + attached_image: Union[Image, str] + num_comments: int + tags: List[str] + vote_state: int + edited: bool + link: str + rt: int + rc: int + user_id: int + user_username: str + user_score: int + user_avatar: UserAvatar + editable: bool + +class Comment(TypedDict): + id: int + rant_id: int + body: str + score: int + created_time: int + vote_state: int + user_id: int + user_username: str + user_score: int + user_avatar: UserAvatar + +class UserProfile(TypedDict): + username: str + score: int + about: str + location: str + created_time: int + skills: str + github: str + website: str + avatar: UserAvatar + content: Dict[str, Dict[str, Union[List[Rant], List[Comment]]]] + +class Notification(TypedDict): + type: str + rant_id: int + comment_id: int + created_time: int + read: int + uid: int # User ID of the notifier + username: str + +# --- API Class --- + class Api: + """An asynchronous wrapper for the devRant API.""" - base_url = "https://www.devrant.io/api/" + base_url: str = "https://www.devrant.io/api/" - def __init__(self, username=None, password=None): - self.username = username - self.password = password - self.auth = None - self.app_id = 3 - self.user_id = None - self.token_id = None - self.token_Key = None - self.session = None + def __init__(self, username: Optional[str] = None, password: Optional[str] = None): + """ + Initializes the API client. - def patch_auth(self, request_dict=None): - auth_dict = {"app": self.app_id} + Args: + username (Optional[str]): The username for authentication. + password (Optional[str]): The password for authentication. + """ + self.username: Optional[str] = username + self.password: Optional[str] = password + self.auth: Optional[AuthToken] = None + self.app_id: int = 3 + self.user_id: Optional[int] = None + self.token_id: Optional[int] = None + self.token_key: Optional[str] = None + self.session: Optional[aiohttp.ClientSession] = None + + def patch_auth(self, request_dict: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """ + Adds authentication details to a request dictionary. + + Args: + request_dict (Optional[Dict[str, Any]]): The dictionary to patch. + + Returns: + Dict[str, Any]: The patched dictionary with auth details. + """ + auth_dict: Dict[str, Any] = {"app": self.app_id} if self.auth: auth_dict.update( user_id=self.user_id, token_id=self.token_id, token_key=self.token_key @@ -32,12 +128,40 @@ class Api: request_dict.update(auth_dict) return request_dict - def patch_url(self, url: str): + def patch_url(self, url: str) -> str: + """ + Constructs the full API URL for an endpoint. + + Args: + url (str): The endpoint path. + + Returns: + str: The full API URL. + """ return self.base_url.rstrip("/") + "/" + url.lstrip("/") - async def login(self): + async def login(self) -> bool: + """ + Authenticates the user and stores the auth token. + + Returns: + bool: True if login is successful, False otherwise. + + Response Structure: + ```json + { + "success": true, + "auth_token": { + "id": int, // Token ID + "key": "string", // Token key + "expire_time": int, // Unix timestamp of token expiration + "user_id": int // ID of the authenticated user + } + } + ``` + """ if not self.username or not self.password: - raise Exception("No authentication defails supplied.") + raise Exception("No authentication details supplied.") async with self as session: response = await session.post( url=self.patch_url("users/auth-token"), @@ -47,7 +171,7 @@ class Api: "app": self.app_id, }, ) - obj = await response.json() + obj: LoginResponse = await response.json() if not obj.get("success"): return False self.auth = obj.get("auth_token") @@ -56,22 +180,46 @@ class Api: self.user_id = self.auth.get("user_id") self.token_id = self.auth.get("id") self.token_key = self.auth.get("key") - return self.auth and True or False + return bool(self.auth) - async def ensure_login(self): + async def ensure_login(self) -> bool: + """Ensures the user is logged in before making a request.""" if not self.auth: return await self.login() return True - async def __aenter__(self): + async def __aenter__(self) -> aiohttp.ClientSession: + """Asynchronous context manager entry.""" self.session = aiohttp.ClientSession() return self.session - async def __aexit__(self, *args, **kwargs): - await self.session.close() + async def __aexit__(self, *args: Any, **kwargs: Any) -> None: + """Asynchronous context manager exit.""" + if self.session and not self.session.closed: + await self.session.close() self.session = None - async def register_user(self, email, username, password): + async def register_user(self, email: str, username: str, password: str) -> bool: + """ + Registers a new user. + + Args: + email (str): The user's email address. + username (str): The desired username. + password (str): The desired password. + + Returns: + bool: True on successful registration, False otherwise. + + Failure Response Structure: + ```json + { + "success": false, + "error": "Error message string.", + "error_field": "field_name" // e.g., "username" or "email" + } + ``` + """ response = None async with self as session: response = await session.post( @@ -88,13 +236,37 @@ class Api: obj = await response.json() return obj.get('success', False) - async def get_comments_from_user(self, username): + async def get_comments_from_user(self, username: str) -> List[Comment]: + """ + Fetches all comments posted by a specific user by first fetching their profile. + + Args: + username (str): The username of the user. + + Returns: + List[Comment]: A list of comment objects. The structure of each comment + is inferred from the general API design. + """ user_id = await self.get_user_id(username) + if not user_id: + return [] profile = await self.get_profile(user_id) + if not profile: + return [] + # The API nests content twice return profile.get("content", {}).get("content", {}).get("comments", []) - async def post_comment(self, rant_id, comment): - response = None + async def post_comment(self, rant_id: int, comment: str) -> bool: + """ + Posts a comment on a specific rant. + + Args: + rant_id (int): The ID of the rant to comment on. + comment (str): The content of the comment. + + Returns: + bool: True if the comment was posted successfully, False otherwise. + """ if not await self.ensure_login(): return False async with self as session: @@ -105,11 +277,19 @@ class Api: obj = await response.json() return obj.get("success", False) - async def get_comment(self, id_): - response = None + async def get_comment(self, id_: int) -> Optional[Comment]: + """ + Retrieves a single comment by its ID. + + Args: + id_ (int): The ID of the comment. + + Returns: + Optional[Comment]: A dictionary representing the comment, or None if not found. + """ async with self as session: response = await session.get( - url=self.patch_url("comments/" + str(id_)), params=self.patch_auth() + url=self.patch_url(f"comments/{id_}"), params=self.patch_auth() ) obj = await response.json() @@ -118,19 +298,58 @@ class Api: return obj.get("comment") - async def delete_comment(self, id_): - response = None + async def delete_comment(self, id_: int) -> bool: + """ + Deletes a comment by its ID. + + Args: + id_ (int): The ID of the comment to delete. + + Returns: + bool: True if deletion was successful, False otherwise. + """ if not await self.ensure_login(): return False async with self as session: response = await session.delete( - url=self.patch_url("comments/" + str(id_)), params=self.patch_auth() + url=self.patch_url(f"comments/{id_}"), params=self.patch_auth() ) obj = await response.json() return obj.get("success", False) - async def get_profile(self, id_): - response = None + async def get_profile(self, id_: int) -> Optional[UserProfile]: + """ + Retrieves the profile of a user by their ID. + + Args: + id_ (int): The user's ID. + + Returns: + Optional[UserProfile]: A dictionary with the user's profile data. + + Profile Structure: + ```json + { + "username": "string", + "score": int, + "about": "string", + "location": "string", + "created_time": int, + "skills": "string", + "github": "string", + "website": "string", + "avatar": { "b": "hex_color", "i": "image_id" }, + "content": { + "content": { + "rants": [ RantObject, ... ], + "upvoted": [ RantObject, ... ], + "comments": [ CommentObject, ... ], + "favorites": [ RantObject, ... ] + } + } + } + ``` + """ async with self as session: response = await session.get( url=self.patch_url(f"users/{id_}"), params=self.patch_auth() @@ -140,7 +359,16 @@ class Api: return None return obj.get("profile") - async def search(self, term): + async def search(self, term: str) -> List[Rant]: + """ + Searches for rants based on a search term. + + Args: + term (str): The term to search for. + + Returns: + List[Rant]: A list of rant objects from the search results. + """ async with self as session: response = await session.get( url=self.patch_url("devrant/search"), @@ -148,11 +376,29 @@ class Api: ) obj = await response.json() if not obj.get("success"): - return + return [] return obj.get("results", []) - async def get_rant(self, id): - response = None + async def get_rant(self, id: int) -> Dict[str, Any]: + """ + Retrieves a single rant and its comments by ID. + + Args: + id (int): The ID of the rant. + + Returns: + Dict[str, Any]: The full API response object. + + Response Structure: + ```json + { + "rant": { RantObject }, + "comments": [ CommentObject, ... ], + "success": true, + "subscribed": 0 or 1 + } + ``` + """ async with self as session: response = await session.get( self.patch_url(f"devrant/rants/{id}"), @@ -160,8 +406,18 @@ class Api: ) return await response.json() - async def get_rants(self, sort="recent", limit=20, skip=0): - response = None + async def get_rants(self, sort: str = "recent", limit: int = 20, skip: int = 0) -> List[Rant]: + """ + Fetches a list of rants. + + Args: + sort (str): The sorting method ('recent', 'top', 'algo'). + limit (int): The number of rants to return. + skip (int): The number of rants to skip for pagination. + + Returns: + List[Rant]: A list of rant objects. + """ async with self as session: response = await session.get( url=self.patch_url("devrant/rants"), @@ -169,11 +425,27 @@ class Api: ) obj = await response.json() if not obj.get("success"): - return + return [] return obj.get("rants", []) - async def get_user_id(self, username): - response = None + async def get_user_id(self, username: str) -> Optional[int]: + """ + Retrieves a user's ID from their username. + + Args: + username (str): The username to look up. + + Returns: + Optional[int]: The user's ID, or None if not found. + + Response Structure: + ```json + { + "success": true, + "user_id": int + } + ``` + """ async with self as session: response = await session.get( url=self.patch_url("get-user-id"), @@ -185,15 +457,31 @@ class Api: return obj.get("user_id") @property - async def mentions(self): + async def mentions(self) -> List[Notification]: + """ + Fetches notifications where the user was mentioned. + + Returns: + List[Notification]: A list of mention notification objects. + """ + notifications = await self.notifs return [ - notif for notif in (await self.notifs) if notif["type"] == "comment_mention" + notif for notif in notifications if notif.get("type") == "comment_mention" ] - async def update_comment(self, comment_id, comment): - response = None + async def update_comment(self, comment_id: int, comment: str) -> bool: + """ + Updates an existing comment. + + Args: + comment_id (int): The ID of the comment to update. + comment (str): The new content of the comment. + + Returns: + bool: True if the update was successful, False otherwise. + """ if not await self.ensure_login(): - return None + return False async with self as session: response = await session.post( url=self.patch_url(f"comments/{comment_id}"), @@ -202,9 +490,20 @@ class Api: obj = await response.json() return obj.get("success", False) - async def vote_rant(self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None): + async def vote_rant(self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None) -> bool: + """ + Casts a vote on a rant. + + Args: + rant_id (int): The ID of the rant to vote on. + vote (Literal[-1, 0, 1]): -1 for downvote, 0 to unvote, 1 for upvote. + reason (Optional[VoteReason]): The reason for a downvote. + + Returns: + bool: True if the vote was successful, False otherwise. + """ if not await self.ensure_login(): - return None + return False async with self as session: response = await session.post( url=self.patch_url(f"devrant/rants/{rant_id}/vote"), @@ -213,9 +512,20 @@ class Api: obj = await response.json() return obj.get("success", False) - async def vote_comment(self, comment_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None): + async def vote_comment(self, comment_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None) -> bool: + """ + Casts a vote on a comment. + + Args: + comment_id (int): The ID of the comment to vote on. + vote (Literal[-1, 0, 1]): -1 for downvote, 0 to unvote, 1 for upvote. + reason (Optional[VoteReason]): The reason for a downvote. + + Returns: + bool: True if the vote was successful, False otherwise. + """ if not await self.ensure_login(): - return None + return False async with self as session: response = await session.post( url=self.patch_url(f"comments/{comment_id}/vote"), @@ -225,12 +535,35 @@ class Api: return obj.get("success", False) @property - async def notifs(self): - response = None + async def notifs(self) -> List[Notification]: + """ + Fetches the user's notification feed. + + Returns: + List[Notification]: A list of notification items. + + Response Structure: + ```json + { + "success": true, + "data": { + "items": [ NotificationObject, ... ], + "check_time": int, // Timestamp of the check + "username_map": [], // Deprecated or unused + "unread": { + "all": int, "upvotes": int, "mentions": int, + "comments": int, "subs": int, "total": int + }, + "num_unread": int + } + } + ``` + """ if not await self.ensure_login(): - return + return [] async with self as session: response = await session.get( url=self.patch_url("users/me/notif-feed"), params=self.patch_auth() ) - return (await response.json()).get("data", {}).get("items", []) + obj = await response.json() + return obj.get("data", {}).get("items", [])