|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any, Optional
|
|
|
|
|
|
class DevRant:
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
username: Optional[str] = None,
|
|
password: Optional[str] = None,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.username = username
|
|
self.password = password
|
|
self.auth: dict[str, Any] = {}
|
|
|
|
def _url(self, path: str) -> str:
|
|
return f"{self.base_url}/api/{path.lstrip('/')}"
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
params: Optional[dict] = None,
|
|
body: Optional[dict] = None,
|
|
) -> dict:
|
|
merged = dict(params or {})
|
|
merged.update(self.auth)
|
|
url = self._url(path)
|
|
data = None
|
|
headers = {"Accept": "application/json"}
|
|
if method in ("GET", "DELETE"):
|
|
if merged:
|
|
url += "?" + urllib.parse.urlencode(merged)
|
|
else:
|
|
payload = dict(merged)
|
|
payload.update(body or {})
|
|
data = urllib.parse.urlencode(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
|
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
with urllib.request.urlopen(request) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
def login(self) -> dict:
|
|
result = self._request(
|
|
"POST",
|
|
"users/auth-token",
|
|
body={"username": self.username, "password": self.password},
|
|
)
|
|
if not result.get("success"):
|
|
raise RuntimeError(result.get("error", "login failed"))
|
|
token = result["auth_token"]
|
|
self.auth = {
|
|
"user_id": token["user_id"],
|
|
"token_id": token["id"],
|
|
"token_key": token["key"],
|
|
}
|
|
return self.auth
|
|
|
|
def register(self, email: str) -> dict:
|
|
return self._request(
|
|
"POST",
|
|
"users",
|
|
body={
|
|
"username": self.username,
|
|
"email": email,
|
|
"password": self.password,
|
|
},
|
|
)
|
|
|
|
def rants(self, sort: str = "recent", limit: int = 20, skip: int = 0) -> list:
|
|
result = self._request(
|
|
"GET", "devrant/rants", {"sort": sort, "limit": limit, "skip": skip}
|
|
)
|
|
return result.get("rants", [])
|
|
|
|
def rant(self, rant_id: int) -> dict:
|
|
return self._request("GET", f"devrant/rants/{rant_id}")
|
|
|
|
def search(self, term: str) -> list:
|
|
return self._request("GET", "devrant/search", {"term": term}).get("results", [])
|
|
|
|
def user_id(self, username: str) -> Optional[int]:
|
|
return self._request("GET", "get-user-id", {"username": username}).get("user_id")
|
|
|
|
def profile(self, user_id: int) -> Optional[dict]:
|
|
return self._request("GET", f"users/{user_id}").get("profile")
|
|
|
|
def notifs(self) -> dict:
|
|
return self._request("GET", "users/me/notif-feed").get("data", {})
|
|
|
|
def clear_notifs(self) -> dict:
|
|
return self._request("DELETE", "users/me/notif-feed")
|
|
|
|
def edit_profile(self, **fields: str) -> dict:
|
|
return self._request("POST", "users/me/edit-profile", body=fields)
|
|
|
|
def post_rant(self, text: str, tags: str = "") -> dict:
|
|
return self._request("POST", "devrant/rants", body={"rant": text, "tags": tags})
|
|
|
|
def edit_rant(self, rant_id: int, text: str, tags: str = "") -> dict:
|
|
return self._request(
|
|
"POST", f"devrant/rants/{rant_id}", body={"rant": text, "tags": tags}
|
|
)
|
|
|
|
def delete_rant(self, rant_id: int) -> dict:
|
|
return self._request("DELETE", f"devrant/rants/{rant_id}")
|
|
|
|
def vote_rant(self, rant_id: int, vote: int) -> dict:
|
|
return self._request(
|
|
"POST", f"devrant/rants/{rant_id}/vote", body={"vote": vote}
|
|
)
|
|
|
|
def favorite(self, rant_id: int) -> dict:
|
|
return self._request("POST", f"devrant/rants/{rant_id}/favorite")
|
|
|
|
def unfavorite(self, rant_id: int) -> dict:
|
|
return self._request("POST", f"devrant/rants/{rant_id}/unfavorite")
|
|
|
|
def comment(self, rant_id: int, text: str) -> dict:
|
|
return self._request(
|
|
"POST", f"devrant/rants/{rant_id}/comments", body={"comment": text}
|
|
)
|
|
|
|
def vote_comment(self, comment_id: int, vote: int) -> dict:
|
|
return self._request("POST", f"comments/{comment_id}/vote", body={"vote": vote})
|
|
|
|
|
|
def from_env() -> DevRant:
|
|
return DevRant(
|
|
os.environ.get("DEVRANT_BASE", "http://localhost:10500"),
|
|
os.environ.get("DEVRANT_USERNAME"),
|
|
os.environ.get("DEVRANT_PASSWORD"),
|
|
)
|