Typed urllib3.util.retry.Retry

retry.py

from collections import namedtuple
from typing import Iterable, Optional, Tuple, TypeVar

import urllib3

RequestHistory = namedtuple(
    "RequestHistory", ["method", "url", "error", "status", "redirect_location"]
)

RetryType = TypeVar("RetryType", bound="Retry")


class Retry:
    """ Retry configuration.

    Each retry attempt will create a new Retry object with updated values, so
    they can be safely reused.

    Retries can be defined as a default for a pool:

        retries = Retry(connect=5, read=2, redirect=5)
        http = PoolManager(retries=retries)
        response = http.request("GET", "http://example.com/")

    Or per-request (which overrides the default for the pool):

        response = http.request("GET", "http://example.com/", retries=Retry(10))

    Retries can be disabled by passing `False`:

        response = http.request("GET", "http://example.com/", retries=False)

    Errors will be wrapped in :class:`~urllib3.exceptions.MaxRetryError` unless
    retries are disabled, in which case the causing exception will be raised.
    """

    BACKOFF_MAX = 120  # Maximum backoff time.

    DEFAULT_METHOD_WHITELIST = frozenset(
        ["HEAD", "TRACE", "GET", "PUT", "OPTIONS", "DELETE"]
    )
    DEFAULT_REDIRECT_HEADERS_BLACKLIST = frozenset(["Authorization"])
    RETRY_AFTER_STATUS_CODES = frozenset([503, 413, 429])

    def __init__(
        self,
        *,
        total: int = 10,
        connect: Optional[int] = None,
        read: Optional[int] = None,
        redirect: Optional[int] = None,
        status: Optional[int] = None,
        method_whitelist: Iterable[str] = DEFAULT_METHOD_WHITELIST,
        status_forcelist: Optional[Iterable[str]] = None,
        backoff_factor: float = 0,
        raise_on_redirect: bool = True,
        raise_on_status: bool = True,
        history: Optional[Tuple[RequestHistory]] = None,
        respect_retry_after_header: bool = True,
        remove_headers_on_redirect: Iterable[str] = DEFAULT_REDIRECT_HEADERS_BLACKLIST,
    ):
        """
        https://urllib3.readthedocs.io/en/latest/reference/urllib3.util.html#urllib3.util.retry.Retry
        """

        @classmethod
        def from_int(
            cls,
            *,
            retries: Optional[int],
            redirect: bool = True,
            default: Optional[RetryType] = None,
        ) -> RetryType:
            """
            Backwards - compatibility for the old retries format.
            """

        def get_backoff_time() -> float:
            """
            Formula for computing the current backoff
            """

        def get_retry_after(response: urllib3.response.HTTPResponse) -> int:
            """
            Get the value of Retry-After in seconds.
            """

        def increment(
            method: Optional[str] = None,
            url: Optional[str] = None,
            response: Optional[urllib3.response.HTTPResponse] = None,
            error: Optional[Exception] = None,
            _pool: Optional[urllib3.connectionpool.HTTPConnectionPool] = None,
            _stacktrace=None,
        ) -> RetryType:
            """
            Return a new Retry object with incremented retry counters.
            """

        def is_exhausted() -> bool:
            """
            Are we out of retries?
            """

        def is_retry(method, status_code, has_retry_after=False) -> bool:
            """
            Is this method/status code retryable? (Based on whitelists and control
            variables such as the number of total retries to allow, whether to respect
            the Retry-After header, whether this header is present, and whether the
            returned status code is on the list of status codes to be retried upon
            on the presence of the aforementioned header)
            """

        def new(**kw) -> RetryType:
            pass

        def parse_retry_after(retry_after: str) -> int:
            pass

        def sleep(response: Optional[urllib3.response.HTTPResponse] = None) -> None:
            """
            Sleep between retry attempts.
            """

        def sleep_for_retry(
            response: Optional[urllib3.response.HTTPResponse] = None
        ) -> bool:
            pass


# For backwards compatibility (equivalent to pre-v1.9):
Retry.DEFAULT = Retry(total=3)