Class GithubRetry

github/GithubRetry.py:54–234

A Github-specific implementation of `urllib3.Retry` This retries 403 responses if they are retry-able. Github requests are retry-able when the response provides a `"Retry-After"` header, or the content indicates a rate limit error. By default, response codes 403, and 500 up to 599

class GithubRetry(Retry):
    """
    A Github-specific implementation of `urllib3.Retry`

    This retries 403 responses if they are retry-able. Github requests are retry-able when
    the response provides a `"Retry-After"` header, or the content indicates a rate limit error.

    By default, response codes 403, and 500 up to 599 are retried. This can be configured
    via the `status_forcelist` argument.

    By default, all methods defined in `Retry.DEFAULT_ALLOWED_METHODS` are retried, plus GET and POST.
    This can be configured via the `allowed_methods` argument.

    """

    __logger: Logger | None = None

    # used to mock datetime, mock.patch("github.GithubRetry.date") does not work as this
    # references the class, not the module (due to re-exporting in github/__init__.py)
    __datetime = datetime

    def __init__(self, secondary_rate_wait: float = DEFAULT_SECONDARY_RATE_WAIT, **kwargs: Any) -> None:
        """
        :param secondary_rate_wait: seconds to wait before retrying secondary rate limit errors
        :param kwargs: see urllib3.Retry for more arguments
        """
        self.secondary_rate_wait = secondary_rate_wait
        # 403 is too broad to be retried, but GitHub API signals rate limits via 403
        # we retry 403 and look into the response header via Retry.increment
        # to determine if we really retry that 403
        kwargs["status_forcelist"] = kwargs.get("status_forcelist", list(range(500, 600))) + [403]
        kwargs["allowed_methods"] = kwargs.get("allowed_methods", Retry.DEFAULT_ALLOWED_METHODS.union({"GET", "POST"}))
        super().__init__(**kwargs)

    def new(self, **kw: Any) -> Self:
        kw.update(dict(secondary_rate_wait=self.secondary_rate_wait))
        return super().new(**kw)  # type: ignore

    def increment(  # type: ignore[override]
        self,
        method: str | None = None,
        url: str | None = None,
        response: HTTPResponse | None = None,  # type: ignore[override]
        error: Exception | None = None,
        _pool: ConnectionPool | None = None,
        _stacktrace: TracebackType | None = None,
    ) -> Retry:
        if response:
            # we retry 403 only when there is a Retry-After header (indicating it is retry-able)
            # or the body message does imply a rate limit error
            if response.status == 403:
                self.__log(
                    logging.INFO,
                    f"Request {method} {url} failed with {response.status}: {response.reason}",
                )
                if "Retry-After" in response.headers:
                    # Sleeping 'Retry-After' seconds is implemented in urllib3.Retry.sleep() and called by urllib3
                    self.__log(