Implement Rate Limiting, Auth, and Info Errors.

Author: Colinxyz

src/recnetpy/rest/exceptions/__init__.py

@@ -1,4 +1,7 @@
 from .http_error import HTTPError
 from .not_found import NotFound
 from .bad_request import BadRequest
-from .internal_server_error import InternalServerError
+from .internal_server_error import InternalServerError
+from .forbidden import Forbidden
+from .rate_limited import RateLimited
+from .unauthorized import Unauthorized

src/recnetpy/rest/exceptions/bad_request.py

@@ -1,8 +1,15 @@
-class BadRequest(Exception):
+from typing import TYPE_CHECKING
+
+from . import HTTPError
+
+if TYPE_CHECKING:
+    from .. import Response
+
+class BadRequest(HTTPError):
     """
     An exception for a 400 HTTP error.
     """
 
-    def __init__(self):
+    def __init__(self, resp: 'Response'):
         message = "Bad request, make sure your input is valid!"
-        super().__init__(message)
+        super().__init__(resp, message)

src/recnetpy/rest/exceptions/forbidden.py

@@ -0,0 +1,15 @@
+from typing import TYPE_CHECKING
+
+from . import HTTPError
+
+if TYPE_CHECKING:
+    from .. import Response
+
+class Forbidden(HTTPError):
+    """
+    An exception raised for 403 HTTP error.
+    """
+
+    def __init__(self, resp: 'Response') -> None:
+        message = "You do not have permission to access this resource!"
+        super().__init__(resp, message)

src/recnetpy/rest/exceptions/http_error.py

@@ -1,15 +1,13 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .. import Response
+
 class HTTPError(Exception):
     """
     This class represents an error or problem with a request.
     """
-    status: int
-    url: str
-    data: any
-
 
-    def __init__(self, status: int, url: str, data: any) -> None:
-        self.status = status
-        self.url = url
-        self.data = data
-        error_message = f"The request to {url} has failed with code {status}. The following data has been returned: \n {data}"
+    def __init__(self, resp: 'Response', msg: str = "No Info.") -> None:
+        error_message = f"Info: {msg} \nURL: {resp.url} \nStatus: {resp.status}\n Data: \n{resp.data}"
         super().__init__(error_message)

src/recnetpy/rest/exceptions/internal_server_error.py

@@ -1,8 +1,15 @@
-class InternalServerError(Exception):
+from typing import TYPE_CHECKING
+
+from . import HTTPError
+
+if TYPE_CHECKING:
+    from .. import Response
+
+class InternalServerError(HTTPError):
     """
     An exception for a 500 HTTP error.
     """
 
-    def __init__(self):
+    def __init__(self, resp: 'Response'):
         message = "Something went wrong within Rec Room's servers. Make sure your input is valid!"
-        super().__init__(message)
+        super().__init__(resp, message)

src/recnetpy/rest/exceptions/not_found.py

@@ -1,8 +1,15 @@
-class NotFound(Exception):
+from typing import TYPE_CHECKING
+
+from . import HTTPError
+
+if TYPE_CHECKING:
+    from .. import Response
+
+class NotFound(HTTPError):
     """
     An exception for a 404 HTTP error.
     """
 
-    def __init__(self):
+    def __init__(self, resp: 'Response'):
         message = "The data you were looking for can't be found! It either doesn't exist or is private."
-        super().__init__(message)
+        super().__init__(resp, message)

src/recnetpy/rest/exceptions/rate_limited.py

@@ -0,0 +1,18 @@
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+from . import HTTPError
+
+if TYPE_CHECKING:
+    from .. import Response
+
+class RateLimited(HTTPError):
+    """
+    This exception is raised when the a rate limit is encountered. Raised for a 401 with a retry-after header.
+    """
+
+    def __init__(self, resp: 'Response') -> None:
+        time_out = resp.headers.get("retry_after")
+        t = timedelta(time_out)
+        message = f"You're currently being rate limited. Time out expires in {t}."
+        super().__init__(resp, message)

src/recnetpy/rest/exceptions/unauthorized.py

@@ -0,0 +1,15 @@
+from typing import TYPE_CHECKING
+
+from . import HTTPError
+
+if TYPE_CHECKING:
+    from .. import Response
+
+class Unauthorized(Exception):
+    """
+    An exception raised for 401 HTTP error.
+    """
+
+    def __init__(self, resp: 'Response') -> None:
+        message = "You aren't authorized to access this rousource. Please verify you entered the correct api key."
+        super().__init__(resp, message)

src/recnetpy/rest/http_client.py

@@ -1,31 +1,65 @@
 from typing import TYPE_CHECKING, Dict
+from math import floor
 
-from asyncio import Lock
+from asyncio import Lock, get_running_loop, AbstractEventLoop, sleep
 from aiohttp import ClientSession, TCPConnector
 
-from .async_threads import AsyncThreadPool
-from .exceptions import HTTPError, BadRequest, InternalServerError
+from .exceptions import *
 
 if TYPE_CHECKING:
     from .request import Request
     from .response import Response
 
+RATE_LIMIT = 30
+
+def verify_status(resp: Response):
+    match resp.status:
+        case 400:
+            raise BadRequest(resp)
+        case 401:
+            raise Unauthorized(resp)
+        case 403:
+            if resp.headers.get("retry-after"):
+                raise RateLimited(resp)
+            raise Forbidden(resp)
+        case 404:
+            raise NotFound(resp)
+        case 500:
+            raise InternalServerError(resp)
+        case _:
+            raise HTTPError(resp)
+
+
 class HTTPClient:
     """
     This class is responsible for managing the
     client session, and adding requests to the
     thread pool.
     """
-    locks: Dict[str, Lock]
     session: ClientSession
-    thread_pool: AsyncThreadPool
+    api_key: str
+    rate_limit: int
+    remaining_limit: int
+    next_tick: float
+    tick_offset: float
+    __loop: AbstractEventLoop
+    __sleep: Lock
+    
 
-    def __init__(self) -> None:
-        self.locks = {}
-        connector = TCPConnector(limit=200)
+    def __init__(self, api_key: str) -> None:
+        connector = TCPConnector(limit=100)
         self.session = ClientSession(connector=connector)
-        self.thread_pool = AsyncThreadPool(200) #Allows ONLY 200 connections to be processed at any given time.
+        self.__sleep = Lock()
+        self.__loop = get_running_loop()
+        self.api_key = api_key
+        self.rate_limit = RATE_LIMIT
+        self.tick_offset = self.__loop.time() % 1
+        self.reset_limit()
 
+    def reset_limit(self):
+        self.next_tick = floor(self.__loop.time() + 1) + self.tick_offset
+        self.remaining_limit = self.rate_limit
+        
     async def push(self, request: 'Request') -> 'Response':
         """
         Creates a lock unique to the request, and 
@@ -35,29 +69,23 @@ async def push(self, request: 'Request') -> 'Response':
         @param request: The request object to be executed.
         @return: Returns a response object. 
         """
-        lock = self.locks.get(request.bucket)
-        if lock is None:
-            lock = Lock()
-            self.locks[request.bucket] = lock
-        async with lock:
-            await self.thread_pool.submit(request)
-            result = await request.get_result()
-            if result.status == 404:
-                result.data = None
-                return result
-            if result.success: return result
-            match result.status:
-                case 400:
-                    raise BadRequest
-                case 500:
-                    raise InternalServerError
-                case _:
-                    raise HTTPError(result.status, request.url, result.data)
+        async with self.__sleep:
+            t = self.__loop.time()
+            if t >= self.next_tick: self.reset_limit()
+            if self.remaining_limit <= 0:
+                await sleep(self.next_tick - t)
+                self.reset_limit()
+            request.send()
+            self.remaining_limit -= 1
+        resp = await request.get_result()
+        verify_status(resp)
+        return resp
+
+
 
     async def stop(self) -> None:
         """
         Stops the thread pool, and closes the
         underlying client connection.
         """
-        await self.thread_pool.stop()
         await self.session.close()

src/recnetpy/rest/request.py

@@ -1,6 +1,7 @@
-from typing import TypeVar, Dict, Optional, Union, List
+from typing import Dict, Optional, Union, List, Generic, TypeVar
 from aiohttp import ClientSession, ClientResponse
-from .async_threads import ThreadTask
+from asyncio import Future
+
 from .response import Response
 
 async def parse_response(resp: ClientResponse) -> Union[str, Dict, List]:
@@ -14,7 +15,9 @@ async def parse_response(resp: ClientResponse) -> Union[str, Dict, List]:
         return await resp.json()
     return await resp.text()
 
-class Request(ThreadTask[Response]):
+RT = TypeVar('RT')
+
+class Request(Generic[RT]):
     """
     This class encapsulates a request to be executed inside of a
     thread pool.
@@ -26,6 +29,8 @@ class Request(ThreadTask[Response]):
     params: Optional[Dict]
     body: Optional[Dict]
     headers: Optional[Dict]
+    result: Optional[Response]
+    __future: Optional[Future]
 
     def __init__(self, client: ClientSession, method: str, url: str, params: Optional[Dict] = None, body: Optional[Dict] = None, headers: Optional[Dict] = None) -> None:
         super().__init__()
@@ -35,16 +40,18 @@ def __init__(self, client: ClientSession, method: str, url: str, params: Optiona
         self.params = params
         self.body = body
         self.headers = headers
+        self.response = None
+        self.__future = None
 
-    async def run(self) -> Response:
+    def send(self) -> Response:
         """
         This function is to be executed within a thread. It makes a
         request, and parses the response into a custom response object.
 
         @return: A response object containing the fetched data.
         """
         self.attempts = 0
-        return await self.make_request()
+        self.__future = self.make_request()
 
     async def make_request(self) -> Response:
         """
@@ -58,17 +65,22 @@ async def make_request(self) -> Response:
         try:
             async with self.client.request(self.method, self.url, data = self.body, params = self.params, headers = self.headers) as response:
                 data = await parse_response(response)
-                return Response(response.status, response.ok, data)
+                return Response(self.url, response.status, response.ok, response.headers, data)
         except Exception as e:
             self.attempts += 1
             if self.attempts <= 3: return await self.make_request()
             raise e
-
-    @property
-    def bucket(self) -> str:
+        
+    async def get_result(self):
         """
-        Represents the request and its componets as astring.
+        It returns the result of the execution. Blocks if 
+        the tasks underlying lock isn't freed, or returns
+        immediately. May cause unexpected behaviour if not
+        submitted to a pool.
 
-        @return: The request as a string. 
+        @return: Returns the value returned from the run method.
         """
-        return f"{self.url}:{self.params}:{self.body}"
+        if self.__future:
+            self.result = await self.__future
+            self.__future = None
+        return self.result