diff --git a/pythorhead/auth.py b/pythorhead/auth.py index 360e0a2..ab1ab46 100644 --- a/pythorhead/auth.py +++ b/pythorhead/auth.py @@ -1,8 +1,5 @@ from typing import Optional -import requests -from loguru import logger - # Stack Overflow: Creating a singleton in Python # https://stackoverflow.com/q/6760685 @@ -20,19 +17,8 @@ class Authentication(metaclass=Singleton): token: Optional[str] = None api_base_url: Optional[str] = None - def log_in(self, username_or_email: str, password: str) -> bool: - payload = { - "username_or_email": username_or_email, - "password": password, - } - try: - re = requests.post(f"{self.api_base_url}/user/login", json=payload) - self.token = re.json()["jwt"] - - except Exception as err: - logger.error(f"Something went wrong while logging in as {username_or_email}: {err}") - return False - return True + def set_token(self, token: str) -> None: + self.token = token - def log_out(self) -> None: - self.token = None + def set_api_base_url(self, api_base_url: str) -> None: + self.api_base_url = api_base_url diff --git a/pythorhead/lemmy.py b/pythorhead/lemmy.py index 871f7ed..bbf7199 100644 --- a/pythorhead/lemmy.py +++ b/pythorhead/lemmy.py @@ -1,33 +1,33 @@ -from typing import Optional +import logging -import requests -from loguru import logger +from typing import Optional -from pythorhead.auth import Authentication from pythorhead.post import Post +from pythorhead.requestor import Requestor, Request + +logging.basicConfig(format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") class Lemmy: post: Post - _auth: Authentication _known_communities = {} + _requestor: Requestor def __init__(self, api_base_url: str) -> None: - self._auth = Authentication() - self._auth.api_base_url = f"{api_base_url}/api/v3" + self._requestor = Requestor() + self._requestor.set_api_base_url(f"{api_base_url}/api/v3") self.post = Post() def log_in(self, username_or_email: str, password: str) -> bool: - return Authentication().log_in(username_or_email, password) + return self._requestor.log_in(username_or_email, password) def discover_community(self, community_name: str) -> Optional[int]: if community_name in self._known_communities: return self._known_communities[community_name] - try: - req = requests.get(f"{self._auth.api_base_url}/community?name={community_name}") - community_id = req.json()["community_view"]["community"]["id"] + + request = self._requestor.request(Request.GET, "/community", params={"name": community_name}) + + if request is not None: + community_id = request["community_view"]["community"]["id"] self._known_communities[community_name] = community_id - except Exception as err: - logger.error(f"Error when looking up community '{community_name}': {err}") - return - return community_id + return community_id diff --git a/pythorhead/post.py b/pythorhead/post.py index ed0c540..510a84c 100644 --- a/pythorhead/post.py +++ b/pythorhead/post.py @@ -1,21 +1,18 @@ from typing import Any, Literal, Optional, List -import requests -from loguru import logger - -from pythorhead.auth import Authentication from pythorhead.types import FeatureType, ListingType, SortType +from pythorhead.requestor import Requestor, Request class Post: def __init__(self): - self._auth = Authentication() + self._requestor = Requestor() def get( self, post_id: int, comment_id: Optional[int] = None, - ) -> dict: + ) -> Optional[dict]: """ Get a post. @@ -27,18 +24,13 @@ def get( dict: post view """ get_post = { - "auth": self._auth.token, "id": post_id, } if comment_id is not None: get_post["comment_id"] = comment_id - re = requests.get(f"{self._auth.api_base_url}/post", params=get_post) - if not re.ok: - logger.error(f"Error encountered while getting posts: {re.text}") - return {} - return re.json() + return self._requestor.request(Request.GET, "/post", params=get_post) def list( # noqa: A003 self, @@ -66,9 +58,7 @@ def list( # noqa: A003 Returns: list[dict]: list of posts """ - list_post: dict[str, Any] = { - "auth": self._auth.token, - } + list_post: dict = {} if community_id is not None: list_post["community_id"] = community_id @@ -84,12 +74,9 @@ def list( # noqa: A003 list_post["sort"] = sort.value if type_ is not None: list_post["type_"] = type_.value - - re = requests.get(f"{self._auth.api_base_url}/post/list", params=list_post) - if not re.ok: - logger.error(f"Error encountered while getting posts: {re.text}") - return [] - return re.json()["posts"] + if data := self._requestor.request(Request.GET, "/post/list", params=list_post): + return data["posts"] + return [] def create( self, @@ -100,7 +87,7 @@ def create( nsfw: Optional[bool] = None, honeypot: Optional[str] = None, language_id: Optional[int] = None, - ) -> bool: + ) -> Optional[dict]: """ Create a post @@ -114,10 +101,9 @@ def create( language_id (int, optional): Defaults to None. Returns: - bool: True if successful + Optional[dict]: post data if successful """ new_post = { - "auth": self._auth.token, "community_id": community_id, "name": name, } @@ -133,14 +119,9 @@ def create( if language_id is not None: new_post["language_id"] = language_id - re = requests.post(f"{self._auth.api_base_url}/post", json=new_post) - - if not re.ok: - logger.error(f"Error encountered while posting: {re.text}") - - return re.ok + return self._requestor.request(Request.POST, "/post", json=new_post) - def delete(self, post_id: int, deleted: bool) -> bool: + def delete(self, post_id: int, deleted: bool) -> Optional[dict]: """ Deletes / Restore a post @@ -149,19 +130,15 @@ def delete(self, post_id: int, deleted: bool) -> bool: deleted (bool) Returns: - bool: True if successful + Optional[dict]: post data if successful """ delete_post = { - "auth": self._auth.token, "post_id": post_id, "deleted": deleted, } - re = requests.post(f"{self._auth.api_base_url}/post/delete", json=delete_post) - if not re.ok: - logger.error(f"Error encountered while deleting post: {re.text}") - return re.ok + return self._requestor.request(Request.POST, "/post/delete", json=delete_post) - def remove(self, post_id: int, removed: bool, reason: Optional[str] = None) -> bool: + def remove(self, post_id: int, removed: bool, reason: Optional[str] = None) -> Optional[dict]: """ Moderator remove / restore a post. @@ -172,19 +149,16 @@ def remove(self, post_id: int, removed: bool, reason: Optional[str] = None) -> b reason (str, optional): Defaults to None. Returns: - bool: True if successful + Optional[dict]: post data if successful """ remove_post = { - "auth": self._auth.token, "post_id": post_id, "removed": removed, } if reason is not None: remove_post["reason"] = reason - re = requests.post(f"{self._auth.api_base_url}/post/remove", json=remove_post) - if not re.ok: - logger.error(f"Error encountered while removing post: {re.text}") - return re.ok + + return self._requestor.request(Request.POST, "/post/remove", json=remove_post) def edit( self, @@ -194,7 +168,7 @@ def edit( body: Optional[str] = None, nsfw: Optional[bool] = None, language_id: Optional[int] = None, - ) -> bool: + ) -> Optional[dict]: """ Edit a post @@ -208,10 +182,9 @@ def edit( language_id (int, optional): Defaults to None. Returns: - bool: True if successful + Optional[dict]: post data if successful """ - edit_post = { - "auth": self._auth.token, + edit_post: dict[str, Any] = { "post_id": post_id, } if name is not None: @@ -224,12 +197,10 @@ def edit( edit_post["nsfw"] = nsfw if language_id is not None: edit_post["language_id"] = language_id - re = requests.put(f"{self._auth.api_base_url}/post", json=edit_post) - if not re.ok: - logger.error(f"Error encountered while editing post: {re.text}") - return re.ok - def like(self, post_id: int, score: Literal[-1, 0, 1]) -> bool: + return self._requestor.request(Request.PUT, "/post", json=edit_post) + + def like(self, post_id: int, score: Literal[-1, 0, 1]) -> Optional[dict]: """ Like a post @@ -238,19 +209,15 @@ def like(self, post_id: int, score: Literal[-1, 0, 1]) -> bool: score (int) Returns: - bool: True if successful + Optional[dict]: post data if successful """ like_post = { - "auth": self._auth.token, "post_id": post_id, "score": score, } - re = requests.post(f"{self._auth.api_base_url}/post/like", json=like_post) - if not re.ok: - logger.error(f"Error encountered while liking post: {re.text}") - return re.ok + return self._requestor.request(Request.POST, "/post/like", json=like_post) - def save(self, post_id: int, saved: bool) -> bool: + def save(self, post_id: int, saved: bool) -> Optional[dict]: """ Save / Unsave a post @@ -260,19 +227,15 @@ def save(self, post_id: int, saved: bool) -> bool: saved (bool) Returns: - bool: True if successful + Optional[dict]: post data if successful """ save_post = { - "auth": self._auth.token, "post_id": post_id, "save": saved, } - re = requests.put(f"{self._auth.api_base_url}/post/save", json=save_post) - if not re.ok: - logger.error(f"Error encountered while saving post: {re.text}") - return re.ok + return self._requestor.request(Request.PUT, "/post/save", json=save_post) - def report(self, post_id: int, reason: str) -> bool: + def report(self, post_id: int, reason: str) -> Optional[dict]: """ Report a post @@ -282,42 +245,33 @@ def report(self, post_id: int, reason: str) -> bool: reason (str) Returns: - bool: True if successful + Optional[dict]: post report data if successful """ report_post = { - "auth": self._auth.token, "post_id": post_id, "reason": reason, } + return self._requestor.request(Request.POST, "/post/report", json=report_post) - re = requests.post(f"{self._auth.api_base_url}/post/report", json=report_post) - if not re.ok: - logger.error(f"Error encountered while reporting post: {re.text}") - return re.ok - - def feature(self, post_id: int, feature: bool, feature_type: FeatureType) -> bool: + def feature(self, post_id: int, feature: bool, feature_type: FeatureType) -> Optional[dict]: """ Add / Remove Feature from a post Args: - post_id (int): - feature (bool): + post_id (int) + feature (bool) feature_type (FeatureType) Returns: - bool: True if successful + Optional[dict]: post data if successful """ feature_post = { - "auth": self._auth.token, "post_id": post_id, "featured": feature, "feature_type": feature_type.value, } - re = requests.post(f"{self._auth.api_base_url}/post/feature", json=feature_post) - if not re.ok: - logger.error(f"Error encountered while feature post: {re.text}") - return re.ok + return self._requestor.request(Request.POST, "/post/feature", json=feature_post) __call__ = create diff --git a/pythorhead/requestor.py b/pythorhead/requestor.py new file mode 100644 index 0000000..c06b788 --- /dev/null +++ b/pythorhead/requestor.py @@ -0,0 +1,56 @@ +import requests +import logging + +from enum import Enum +from typing import Optional +from pythorhead.auth import Authentication + + +logger = logging.getLogger(__name__) + + +class Request(Enum): + GET = "GET" + PUT = "PUT" + POST = "POST" + + +REQUEST_MAP = { + Request.GET: requests.get, + Request.PUT: requests.put, + Request.POST: requests.post, +} + + +class Requestor: + def __init__(self): + self._auth = Authentication() + self.set_api_base_url = self._auth.set_api_base_url + + def request(self, method: Request, endpoint: str, **kwargs) -> Optional[dict]: + logger.info(f"Requesting {method} {endpoint}") + if self._auth.token: + if (data := kwargs.get("json")) is not None: + data["auth"] = self._auth.token + if (data := kwargs.get("params")) is not None: + data["auth"] = self._auth.token + + r = REQUEST_MAP[method](f"{self._auth.api_base_url}{endpoint}", **kwargs) + + if not r.ok: + logger.error(f"Error encountered while {method}: {r.text}") + return + + return r.json() + + def log_in(self, username_or_email: str, password: str) -> bool: + payload = { + "username_or_email": username_or_email, + "password": password, + } + if data := self.request(Request.POST, "/user/login", json=payload): + self._auth.set_token(data["jwt"]) + return self._auth.token is not None + + def log_out(self) -> None: + self._auth.token = None diff --git a/requirements.txt b/requirements.txt index e7437b0..0bf71d0 100644 Binary files a/requirements.txt and b/requirements.txt differ