diff --git a/.gitignore b/.gitignore index 632e35f..3280530 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,6 @@ __pycache__ dist docs/_build boticordpy.egg-info -test.py +_testing.py /.pytest_cache /docs/build/ diff --git a/boticordpy/__init__.py b/boticordpy/__init__.py index 8157c2b..eee1232 100644 --- a/boticordpy/__init__.py +++ b/boticordpy/__init__.py @@ -1,16 +1,16 @@ """ Boticord API Wrapper ~~~~~~~~~~~~~~~~~~~ -A basic wrapper for the Boticord API. +A basic wrapper for the BotiCord API. :copyright: (c) 2022 Marakarka :license: MIT, see LICENSE for more details. """ -__title__ = 'boticordpy' -__author__ = 'Marakarka' -__license__ = 'MIT' -__copyright__ = 'Copyright 2022 Marakarka' -__version__ = '2.1.0' +__title__ = "boticordpy" +__author__ = "Marakarka" +__license__ = "MIT" +__copyright__ = "Copyright 2022 Marakarka" +__version__ = "2.1.0" from .client import BoticordClient from .webhook import Webhook diff --git a/boticordpy/autopost.py b/boticordpy/autopost.py index 131e292..6c09de9 100644 --- a/boticordpy/autopost.py +++ b/boticordpy/autopost.py @@ -20,7 +20,7 @@ class AutoPost: "_error", "_stats", "_task", - "_stopped" + "_stopped", ) _success: typing.Any @@ -134,7 +134,9 @@ class AutoPost: Boticord recommends not to set interval lower than 900 seconds! """ if seconds < 900: - raise ValueError("no. Boticord recommends not to set interval lower than 900 seconds!") + raise ValueError( + "no. Boticord recommends not to set interval lower than 900 seconds!" + ) self._interval = seconds return self @@ -170,7 +172,9 @@ class AutoPost: raise bexc.InternalException("You must provide stats") if self.is_running: - raise bexc.InternalException("Automatically stats posting is already running") + raise bexc.InternalException( + "Automatically stats posting is already running" + ) task = asyncio.ensure_future(self._internal_loop()) self._task = task @@ -178,7 +182,7 @@ class AutoPost: def stop(self) -> None: """ - Stops the autopost. + Stops the autopost. """ if not self.is_running: return None diff --git a/boticordpy/client.py b/boticordpy/client.py index 244721b..e8c5ca0 100644 --- a/boticordpy/client.py +++ b/boticordpy/client.py @@ -15,11 +15,8 @@ class BoticordClient: token (:obj:`str`) Your bot's Boticord API Token. """ - __slots__ = ( - "http", - "_autopost", - "_token" - ) + + __slots__ = ("http", "_autopost", "_token") http: HttpClient @@ -56,7 +53,9 @@ class BoticordClient: response = await self.http.get_bot_comments(bot_id) return [boticord_types.SingleComment(**comment) for comment in response] - async def post_bot_stats(self, servers: int = 0, shards: int = 0, users: int = 0) -> dict: + async def post_bot_stats( + self, servers: int = 0, shards: int = 0, users: int = 0 + ) -> dict: """Post Bot's stats. Args: @@ -70,11 +69,9 @@ class BoticordClient: :obj:`dict`: Boticord API Response status """ - response = await self.http.post_bot_stats({ - "servers": servers, - "shards": shards, - "users": users - }) + response = await self.http.post_bot_stats( + {"servers": servers, "shards": shards, "users": users} + ) return response async def get_server_info(self, server_id: int) -> boticord_types.Server: @@ -160,6 +157,61 @@ class BoticordClient: response = await self.http.get_user_bots(user_id) return [boticord_types.SimpleBot(**bot) for bot in response] + async def get_my_shorted_links(self, *, code: str = None): + """Gets shorted links of an authorized user + + Args: + code (:obj:`str`) + Code of shorted link. Could be None. + + Returns: + Union[:obj:`list` [ :obj:`~.types.ShortedLink` ], :obj:`~types.ShortedLink`]: + List of shorted links if none else shorted link + """ + response = await self.http.get_my_shorted_links(code) + + return ( + [boticord_types.ShortedLink(**link) for link in response] + if code is None + else boticord_types.ShortedLink(**response[0]) + ) + + async def create_shorted_link(self, *, code: str, link: str, domain: boticord_types.LinkDomain = 1): + """Creates new shorted link + + Args: + code (:obj:`str`) + Code of link to short. + link (:obj:`str`) + Link to short. + domain (:obj:`~.types.LinkDomain`) + Domain to use in shorted link + + Returns: + :obj:`~types.ShortedLink`: + Shorted Link + """ + response = await self.http.create_shorted_link(code, link, domain=domain) + + return boticord_types.ShortedLink(**response) + + async def delete_shorted_link(self, code: str, domain: boticord_types.LinkDomain = 1): + """Deletes shorted link + + Args: + code (:obj:`str`) + Code of link to delete. + domain (:obj:`~.types.LinkDomain`) + Domain that is used in shorted link + + Returns: + :obj:`bool`: + Is link deleted successfully? + """ + response = await self.http.delete_shorted_link(code, domain) + + return response.get('ok', False) + def autopost(self) -> AutoPost: """Returns a helper instance for auto-posting. diff --git a/boticordpy/http.py b/boticordpy/http.py index 504224e..515fc27 100644 --- a/boticordpy/http.py +++ b/boticordpy/http.py @@ -3,6 +3,7 @@ import asyncio import aiohttp from . import exceptions +from .types import LinkDomain class HttpClient: @@ -23,26 +24,21 @@ class HttpClient: self.token = auth_token self.API_URL = "https://api.boticord.top/v1/" - loop = kwargs.get('loop') or asyncio.get_event_loop() + loop = kwargs.get("loop") or asyncio.get_event_loop() - self.session = kwargs.get('session') or aiohttp.ClientSession(loop=loop) + self.session = kwargs.get("session") or aiohttp.ClientSession(loop=loop) - async def make_request(self, - method: str, - endpoint: str, - **kwargs): + async def make_request(self, method: str, endpoint: str, **kwargs): """Send requests to the API""" kwargs["headers"] = { "Content-Type": "application/json", - "Authorization": self.token + "Authorization": self.token, } url = f"{self.API_URL}{endpoint}" - async with self.session.request(method, - url, - **kwargs) as response: + async with self.session.request(method, url, **kwargs) as response: data = await response.json() if response.status == 200: @@ -97,3 +93,21 @@ class HttpClient: def get_user_bots(self, user_id: int): """Get bots of specified user""" return self.make_request("GET", f"bots/{user_id}") + + def get_my_shorted_links(self, code: str = None): + """Get shorted links of an authorized user""" + body = {"code": code} if code is not None else {} + + return self.make_request("POST", "links/get", json=body) + + def create_shorted_link(self, code: str, link: str, *, domain: LinkDomain = 1): + """Create new shorted link""" + return self.make_request( + "POST", "links/create", json={"code": code, "link": link, "domain": int(domain)} + ) + + def delete_shorted_link(self, code: str, domain: LinkDomain = 1): + """Delete shorted link""" + return self.make_request( + "POST", "links/delete", json={"code": code, "domain": int(domain)} + ) diff --git a/boticordpy/types.py b/boticordpy/types.py index db91f92..02d4210 100644 --- a/boticordpy/types.py +++ b/boticordpy/types.py @@ -1,4 +1,5 @@ import typing +from enum import IntEnum KT = typing.TypeVar("KT") VT = typing.TypeVar("VT") @@ -59,8 +60,7 @@ def parse_user_comments_dict(response_data: dict) -> dict: class ApiData(dict, typing.MutableMapping[KT, VT]): - """Base class used to represent received data from the API. - """ + """Base class used to represent received data from the API.""" def __init__(self, **kwargs: VT) -> None: super().__init__(**parse_response_dict(kwargs)) @@ -94,6 +94,7 @@ class SingleComment(ApiData): class Bot(ApiData): """This model represents a bot, returned from the BotiCord API""" + id: str """Bot's Id""" @@ -340,3 +341,38 @@ class CommentResponse(ApiData): def __init__(self, **kwargs): super().__init__(**parse_webhook_response_dict(kwargs)) + +class LinkDomain(IntEnum): + """Domain to short the link""" + + BCORD_CC = 1 + """``bcord.cc`` domain, default""" + + MYSERVERS_ME = 2 + """``myservers.me`` domain""" + + DISCORD_CAMP = 3 + """``discord.camp`` domain""" + + +class ShortedLink(ApiData): + id: int + """Id of shorted link""" + + code: str + """Code of shorted link""" + + owner_i_d: str + """Id of owner of shorted link""" + + domain: str + """Domain of shorted link""" + + views: int + """Link views count""" + + date: int + """Timestamp of link creation moment""" + + def __init__(self, **kwargs): + super().__init__(**parse_response_dict(kwargs)) diff --git a/boticordpy/webhook.py b/boticordpy/webhook.py index f6bbcc0..49aa084 100644 --- a/boticordpy/webhook.py +++ b/boticordpy/webhook.py @@ -20,6 +20,7 @@ class Webhook: Keyword Arguments: loop: `asyncio loop` """ + __slots__ = ( "_webserver", "_listeners", @@ -27,7 +28,7 @@ class Webhook: "__app", "_endpoint_name", "_x_hook_key", - "_loop" + "_loop", ) __app: web.Application @@ -39,7 +40,7 @@ class Webhook: self._listeners = {} self.__app = web.Application() self._is_running = False - self._loop = kwargs.get('loop') or asyncio.get_event_loop() + self._loop = kwargs.get("loop") or asyncio.get_event_loop() def listener(self, response_type: str): """Decorator to set the listener. @@ -47,6 +48,7 @@ class Webhook: response_type (:obj:`str`) Type of response (Check reference page) """ + def inner(func): if not asyncio.iscoroutinefunction(func): raise TypeError(f"<{func.__qualname__}> must be a coroutine function") @@ -80,7 +82,11 @@ class Webhook: if responder is not None: await responder( - (BumpResponse if data["type"].endswith("_bump") else CommentResponse)(**data) + ( + BumpResponse + if data["type"].endswith("_bump") + else CommentResponse + )(**data) ) return web.Response(status=200)