Skip to content

Commit

Permalink
created requestor class, using standard logging now
Browse files Browse the repository at this point in the history
  • Loading branch information
NicKoehler committed Jun 19, 2023
1 parent aeac389 commit 3cb5a89
Show file tree
Hide file tree
Showing 5 changed files with 112 additions and 116 deletions.
22 changes: 4 additions & 18 deletions pythorhead/auth.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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
30 changes: 15 additions & 15 deletions pythorhead/lemmy.py
Original file line number Diff line number Diff line change
@@ -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
120 changes: 37 additions & 83 deletions pythorhead/post.py
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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
Expand All @@ -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,
}
Expand All @@ -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
Expand All @@ -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.
Expand All @@ -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,
Expand All @@ -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
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Loading

0 comments on commit 3cb5a89

Please sign in to comment.