Advanced¶
Adding Custom Client¶
The module is written to be as easily portable as possible. Therefore, any custom
client should be easily written to this library. The yippi.AbstractYippi.AbstractYippi
class should be enough for you to inherit. And what you need to do is override the abstract functions.
For example, here’s a snippet from AsyncYippiClient.
from typing import List
from typing import Optional
from typing import Union
import aiohttp
from aiohttp import BasicAuth
from aiohttp import FormData
from .AbstractYippi import AbstractYippi
from .AbstractYippi import limiter
from .Classes import Flag
from .Classes import Note
from .Classes import Pool
from .Classes import Post
from .Exceptions import APIError
from .Exceptions import UserError
class AsyncYippiClient(AbstractYippi):
def __init__(
self,
project_name: str,
version: str,
creator: str,
loop=None,
session: aiohttp.ClientSession = None,
) -> None:
self._loop = loop
self._session: aiohttp.ClientSession = session or aiohttp.ClientSession()
super().__init__(project_name, version, creator)
async def close(self) -> None:
await self._session.close()
async def __aenter__(self) -> "AsyncYippiClient":
return self
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
await self.close()
@limiter.ratelimit("call_api", delay=True)
async def _call_api(
self,
method: str,
url: str,
data: Union[dict, FormData] = None,
file=None,
**kwargs
) -> Optional[Union[List[dict], dict]]:
auth = None
if self._login != ("", ""):
auth = BasicAuth(*self._login)
if file:
file = file["upload[file]"]
formdata = FormData()
formdata.add_field(
"upload[file]", file[1], filename=file[0], content_type=file[2]
)
formdata.add_fields(data)
data = formdata
query_string = self._generate_query_keys(**kwargs)
url += "?" + query_string
r = await self._session.request(
method, url, data=data, headers=self.headers, auth=auth
)
await self._verify_response(r)
if not r.status == 204:
return await r.json()
return None
async def _verify_response(self, r) -> None:
if 300 <= r.status < 500:
res = await r.json()
if r.status >= 400:
raise UserError(res.get("message") or res.get("reason"), json=res)
elif r.status >= 500:
raise APIError(r.reason)
if r.status != 204 and (
not r.headers.get("Content-Type")
or "application/json" not in r.headers.get("Content-Type")
):
res = await r.text()
if "Not found." in res:
raise UserError("Not found.")
raise UserError("Invalid input or server error.")
async def posts(
self,
tags: Union[List, str] = None,
limit: int = None,
page: Union[int, str] = None,
) -> List[Post]:
response = await self._get_posts(tags, limit, page) # type: ignore
posts = [Post(p, client=self) for p in response["posts"]]
return posts
async def post(self, post_id: int) -> Post:
api_res = await self._get_post(post_id) # type: ignore
return Post(api_res["post"], client=self)
async def notes(
self,
body_matches: str = None,
post_id: int = None,
post_tags_match: Union[List, str] = None,
creator_name: str = None,
creator_id: int = None,
is_active: bool = None,
limit: int = None,
) -> List[Note]:
response = await self._get_notes(
body_matches,
post_id,
post_tags_match,
creator_name,
creator_id,
is_active,
limit,
) # type: ignore
result = [Note(n, client=self) for n in response]
return result
async def flags(
self,
post_id: int = None,
creator_id: int = None,
creator_name: str = None,
limit: int = None,
) -> List[Flag]:
response = await self._get_flags(post_id, creator_id, creator_name) # type: ignore
result = [Flag(f, client=self) for f in response]
return result
async def pools(
self,
name_matches: str = None,
id_: Union[int, List[int]] = None,
description_matches: str = None,
creator_name: str = None,
creator_id: int = None,
is_active: bool = None,
is_deleted: bool = None,
category: str = None,
order: str = None,
limit: int = None,
) -> List[Pool]:
response = await self._get_pools(
name_matches,
id_,
description_matches,
creator_name,
creator_id,
is_active,
is_deleted,
category,
order,
limit,
) # type: ignore
result = [Pool(p, client=self) for p in response]
return result
async def pool(self, pool_id: int) -> Pool:
response = await self._get_pool(pool_id) # type: ignore
return Pool(response, client=self)