corbz/PYRSS-Bot
corbz/PYRSS-Bot
1
0
Fork 0
Code Issues 14 Pull Requests Actions Releases 2 Activity

API code improvements + docstrings

Browse Source
This commit is contained in:
Corban-Lee Jones 2024-02-06 23:49:31 +00:00
parent 9031cc90c9
commit fb432631d3
1 changed files with 72 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
src/api.py
View File

@ -31,79 +31,97 @@ class API:
self.session = session self.session = session
self.token_headers = {"Authorization": f"Token {api_token}"} self.token_headers = {"Authorization": f"Token {api_token}"}
async def fetch_data(self, url: str, params=None): async def make_request(self, method: str, url: str, **kwargs) -> dict:
log.debug("api fetching from url: %s", url) """Make a request to the given API endpoint.
async with self.session.get(url, params=params, headers=self.token_headers) as response:
return await response.json(), await response.text(), response.status
async def send_data(self, url: str, data: dict): Args:
log.debug("api posting to url: %s", url) method (str): The request method to use, examples: GET, POST, DELETE...
async with self.session.post(url, data=data, headers=self.token_headers) as response: url (str): The API endpoint to request to.
return await response.json(), await response.text(), response.status **kwargs: Passed into self.session.request.
async def delete_data(self, url: str): Returns:
log.debug("api deleting to url %s", url) dict: Dictionary containing status code, json or text.
async with self.session.delete(url, headers=self.token_headers) as response: """
return await response.text(), response.status
async def create_new_rssfeed(self, name: str, url: str, image_url: str, discord_server_id: int): async with self.session.request(method, url, headers=self.token_headers, **kwargs) as response:
response.raise_for_status()
try:
json = await response.json()
text = None
except aiohttp.ContentTypeError:
json = None
text = await response.text()
log.debug("api creating rss feed: %s %s %s", name, url, image_url) status = response.status
return {"json": json, "text": text, "status": status}
async def create_new_rssfeed(self, name: str, url: str, image_url: str, discord_server_id: int) -> dict:
"""Create a new RSS Feed.
Args:
name (str): Name of the RSS Feed.
url (str): URL for the RSS Feed.
image_url (str): URL of the image representation of the RSS Feed.
discord_server_id (int): ID of the discord server behind this item.
Returns:
dict: JSON representation of the newly created RSS Feed.
"""
log.debug("creating rssfeed: %s %s %s %s", name, url, image_url, discord_server_id)
async with self.session.get(image_url) as response: async with self.session.get(image_url) as response:
image_data = await response.read() image_data = await response.read()
form = aiohttp.FormData() # Using formdata to make the image transfer easier.
form.add_field("name", name) form = aiohttp.FormData({
form.add_field("url", url) "name": name,
"url": url,
"discord_server_id": discord_server_id
})
form.add_field("image", image_data, filename="file.jpg") form.add_field("image", image_data, filename="file.jpg")
form.add_field("discord_server_id", str(discord_server_id))
resp_json, resp_text, resp_status = await self.send_data(self.RSS_FEED_ENDPOINT, form) data = (await self.make_response("POST", self.RSS_FEED_ENDPOINT, data=form))["json"]
return data
if resp_status != 201:
log.error(resp_text)
raise NotCreatedException(f"Expected HTTP 201, not HTTP {response.status} - {resp_text}")
log.debug(resp_text)
return resp_json
async def get_rssfeed(self, uuid: str) -> dict: async def get_rssfeed(self, uuid: str) -> dict:
"""Get a particular RSS Feed given it's UUID.
log.debug("api getting rss feed") Args:
uuid (str): Identifier of the desired RSS Feed.
endpoint = f"{self.RSS_FEED_ENDPOINT}{uuid}" Returns:
resp_json, resp_text, resp_status = await self.fetch_data(endpoint) dict: A JSON representation of the RSS Feed.
"""
if resp_status != 200: log.debug("getting rssfeed: %s", uuid)
log.error(resp_text) endpoint = f"{self.RSS_FEED_ENDPOINT}/{uuid}/"
raise BadStatusException(f"Expected HTTP 200, not HTTP {resp_status} - {resp_text}") data = (await self.make_request("GET", endpoint))["json"]
return data
return resp_json async def get_rssfeed_list(self, **filters) -> tuple[list[dict], int]:
"""Get all RSS Feeds with the associated filters.
async def get_rssfeed_list(self, **filters) -> dict: Returns:
tuple[list[dict], int] list contains dictionaries of each item, int is total items.
"""
log.debug("api getting list of rss feed") log.debug("getting list of rss feeds with filters: %s", filters)
data = (await self.make_request("GET", self.RSS_FEED_ENDPOINT, params=filters))["json"]
return data["results"], data["count"]
resp_json, resp_text, resp_status = await self.fetch_data(self.RSS_FEED_ENDPOINT, params=filters) async def delete_rssfeed(self, uuid: str) -> int:
"""Delete a specified RSS Feed.
if resp_status != 200: Args:
log.error(resp_text) uuid (str): Identifier of the RSS Feed to delete.
raise BadStatusException(f"Expected HTTP 200, not HTTP {resp_status} - {resp_text}")
log.debug(resp_text) Returns:
int: Status code of the response.
"""
return resp_json["results"], resp_json["count"] log.debug("deleting rssfeed: %s", uuid)
endpoint = f"{self.RSS_FEED_ENDPOINT}/{uuid}/"
async def delete_rssfeed(self, uuid: str): status = (await self.make_request("DELETE", endpoint))["status"]
return status
log.debug("api deleting rss feed")
resp_text, resp_status = await self.delete_data(f"{self.RSS_FEED_ENDPOINT}{uuid}/")
if resp_status != 204:
log.error(resp_text)
raise BadStatusException(f"Expected HTTP 204, not HTTP {resp_status} - {resp_text}")
log.debug(resp_text)