add RateLimitExceededError

Author: schelv

githubkit/core.py

@@ -1,3 +1,4 @@
+from datetime import timedelta, datetime, timezone
 from types import TracebackType
 from contextvars import ContextVar
 from contextlib import contextmanager, asynccontextmanager
@@ -22,7 +23,7 @@
 from .response import Response
 from .utils import obj_to_jsonable
 from .config import Config, get_config
-from .exception import RequestError, RequestFailed, RequestTimeout
+from .exception import RequestError, RequestFailed, RequestTimeout, RateLimitExceededError
 from .auth import BaseAuthStrategy, TokenAuthStrategy, UnauthAuthStrategy
 from .typing import (
     URLTypes,
@@ -316,6 +317,21 @@ def _check(
         if response.is_error:
             error_models = error_models or {}
             status_code = str(response.status_code)
+
+            if response.status_code == 403 and "retry-after" in response.headers:
+                raise RateLimitExceededError(timedelta(seconds=int(response.headers["retry-after"])), original_message=response.content)
+
+            if (
+                response.status_code in [403, 429]
+                and "x-ratelimit-remaining" in response.headers
+                and response.headers["x-ratelimit-remaining"] == "0"
+            ):
+                retry_after = datetime.fromtimestamp(
+                    int(response.headers["x-ratelimit-reset"]), tz=timezone.utc
+                ) - datetime.now(tz=timezone.utc)
+                retry_after = max(retry_after, timedelta())
+                raise RateLimitExceededError(retry_after, original_message=response.content)
+
             error_model = error_models.get(
                 status_code,
                 error_models.get(

githubkit/exception.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from typing import TYPE_CHECKING
 
 import httpx
@@ -23,6 +24,20 @@ class RequestError(GitHubException):
     """Simple API request failed with unknown error"""
 
 
+class RateLimitExceededError(GitHubException):
+    """Rate Limit Exceeded"""
+
+    def __init__(self, retry_after: timedelta, original_message=""):
+        self.retry_after = retry_after
+        self.original_message = original_message
+
+    def __str__(self) -> str:
+        error_message = f"You have exceeded a Github API rate limit. Retry after: {self.retry_after}"
+        if self.original_message:
+            error_message += f"\nResponse message from Github:\n{self.original_message}"
+        return error_message
+
+
 class RequestTimeout(GitHubException):
     """Simple API request timeout"""