autopost feature

boticordpy/autopost.py

@@ -0,0 +1,120 @@
+import asyncio
+import typing
+
+from . import exceptions as bexc
+
+
+class AutoPost:
+    __slots__ = (
+        "client",
+        "_interval",
+        "_success",
+        "_error",
+        "_stats",
+        "_task",
+        "_stopped"
+    )
+
+    _success: typing.Any
+    _error: typing.Any
+    _stats: typing.Any
+    _task: typing.Optional["asyncio.Task[None]"]
+
+    def __init__(self, client):
+        self.client = client
+        self._stopped: bool = False
+        self._interval: int = 900
+        self._task: typing.Optional["asyncio.Task[None]"] = None
+
+    @property
+    def is_running(self) -> bool:
+        return self._task is not None and not self._task.done()
+
+    def on_success(self, callback: typing.Any = None):
+        if callback is not None:
+            self._success = callback
+            return self
+
+        def inner(func):
+            if not asyncio.iscoroutinefunction(func):
+                raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
+
+            self._success = callback
+            return func
+
+        return inner
+
+    def on_error(self, callback: typing.Any = None):
+        if callback is not None:
+            self._error = callback
+            return self
+
+        def inner(func):
+            if not asyncio.iscoroutinefunction(func):
+                raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
+
+            self._error = callback
+            return func
+
+        return inner
+
+    def init_stats(self, callback: typing.Any = None):
+        if callback is not None:
+            self._stats = callback
+            return self
+
+        def inner(func):
+            if not asyncio.iscoroutinefunction(func):
+                raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
+
+            self._stats = callback
+            return func
+
+        return inner
+
+    @property
+    def interval(self) -> float:
+        return self._interval
+
+    def set_interval(self, seconds: int) -> "AutoPost":
+        if seconds < 900:
+            raise ValueError("no. Boticord recommends not to set interval lower than 900 seconds!")
+
+        self._interval = seconds
+        return self
+
+    async def _internal_loop(self) -> None:
+        while True:
+            stats = await self._stats()
+            try:
+                await self.client.http.post_bot_stats(stats)
+            except Exception as err:
+                on_error = getattr(self, "_error", None)
+                if on_error:
+                    await on_error(err)
+            else:
+                on_success = getattr(self, "_success", None)
+                if on_success:
+                    await on_success()
+
+            if self._stopped:
+                return None
+
+            await asyncio.sleep(self._interval)
+
+    def start(self):
+        if not hasattr(self, "_stats"):
+            raise bexc.InternalException("You must provide stats")
+
+        if self.is_running:
+            raise bexc.InternalException("Automatically stats posting is already running")
+
+        task = asyncio.ensure_future(self._internal_loop())
+        self._task = task
+        return task
+
+    def stop(self) -> None:
+        if not self.is_running:
+            return None
+
+        self._stopped = True

boticordpy/client.py

@@ -1,50 +1,67 @@
+import typing
+
 from . import types as boticord_types
 from .http import HttpClient
+from .autopost import AutoPost
 
 
 class BoticordClient:
     __slots__ = (
-        "http"
+        "http",
+        "_autopost",
+        "_token"
     )
 
     http: HttpClient
 
     def __init__(self, token=None, **kwargs):
         self._token = token
+        self._autopost: typing.Optional[AutoPost] = None
         self.http = HttpClient(token)
 
-    async def get_bot_info(self, bot_id: int):
+    async def get_bot_info(self, bot_id: int) -> boticord_types.Bot:
         response = await self.http.get_bot_info(bot_id)
         return boticord_types.Bot(**response)
 
-    async def get_bot_comments(self, bot_id: int):
+    async def get_bot_comments(self, bot_id: int) -> list:
         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):
-        response = await self.http.post_bot_stats(servers, shards, users)
+    async def post_bot_stats(self, servers: int = 0, shards: int = 0, users: int = 0) -> dict:
+        response = await self.http.post_bot_stats({
+            "servers": servers,
+            "shards": shards,
+            "users": users
+        })
         return response
 
-    async def get_server_info(self, server_id: int):
+    async def get_server_info(self, server_id: int) -> boticord_types.Server:
         response = await self.http.get_server_info(server_id)
         return boticord_types.Server(**response)
 
-    async def get_server_comments(self, server_id: int):
+    async def get_server_comments(self, server_id: int) -> list:
         response = await self.http.get_server_comments(server_id)
         return [boticord_types.SingleComment(**comment) for comment in response]
 
-    async def post_server_stats(self, payload: dict):
+    async def post_server_stats(self, payload: dict) -> dict:
         response = await self.post_server_stats(payload)
         return response
 
-    async def get_user_info(self, user_id: int):
+    async def get_user_info(self, user_id: int) -> boticord_types.UserProfile:
         response = await self.get_user_info(user_id)
         return boticord_types.UserProfile(**response)
 
-    async def get_user_comments(self, user_id: int):
+    async def get_user_comments(self, user_id: int) -> boticord_types.UserComments:
         response = await self.get_user_comments(user_id)
         return boticord_types.UserComments(**response)
 
-    async def get_user_bots(self, user_id: int):
+    async def get_user_bots(self, user_id: int) -> list:
         response = await self.get_user_bots(user_id)
         return [boticord_types.SimpleBot(**bot) for bot in response]
+
+    def autopost(self) -> AutoPost:
+        if self._autopost is not None:
+            return self._autopost
+
+        self._autopost = AutoPost(self)
+        return self._autopost

boticordpy/exceptions.py

@@ -4,6 +4,19 @@ class BoticordException(Exception):
     """
 
 
+class InternalException(BoticordException):
+    """Exception that's thrown when an local operation operation fails.
+
+    Attributes
+    ----------
+    response:
+        The meaning of the exception
+    """
+
+    def __init__(self, response):
+        super().__init__(response)
+
+
 class HTTPException(BoticordException):
     """Exception that's thrown when an HTTP request operation fails.
 

boticordpy/http.py

@@ -53,15 +53,8 @@ class HttpClient:
     def get_bot_comments(self, bot_id: int):
         return self.make_request("GET", f"bot/{bot_id}/comments")
 
-    def post_bot_stats(self,
-                       servers: int = 0,
-                       shards: int = 0,
-                       users: int = 0):
-        return self.make_request("POST", "stats", json={
-            "servers": servers,
-            "shards": shards,
-            "users": users
-        })
+    def post_bot_stats(self, stats: dict):
+        return self.make_request("POST", "stats", json=stats)
 
     def get_server_info(self, server_id: int):
         return self.make_request("GET", f"server/{server_id}")