Advanced usage of python requests - timeouts, retries, hooks

advanced_requests.py

import requests

from requests.adapters import HTTPAdapter, Retry


DEFAULT_TIMEOUT = 5


class TimeoutHTTPAdapter(HTTPAdapter):
    """Set a default timeout for all HTTP calls."""

    def __init__(self, *args, **kwargs):
        self._timeout = kwargs.pop("timeout", DEFAULT_TIMEOUT)
        super().__init__(*args, **kwargs)

    def send(self, *args, **kwargs):
        kwargs["timeout"] = (
            kwargs["timeout"] if kwargs.get("timeout") else self._timeout
        )
        return super().send(*args, **kwargs)


def assert_status_hook(response, *args, **kwargs):
    """Assert there were no errors with the HTTP request response."""
    response.raise_for_status()


class AdvancedRequest:
    _retries = None
    _timeout_adapter = None
    _session = None

    def __init__(
        self,
        timeout=5,
        retry_total=10,
        retry_allowed_methods=None,
        retry_status_forcelist=None,
        retry_backoff_factor=1,
    ):
        """
        A wrapper for ``requests.Session()`` that also adds a default timeout
        for all HTTP requests, a default retry stratgey with a backoff period
        and a hook to assert the status of every HTTP response.

        References:
        - https://findwork.dev/blog/advanced-usage-python-requests-timeouts-retries-hooks/
        - https://requests.readthedocs.io/en/latest/user/advanced/#timeouts
        - https://urllib3.readthedocs.io/en/latest/reference/urllib3.util.html#urllib3.util.Retry

        :param timeout:
            The **connect** timeout is the number of seconds your client will
            wait to establish a connection to the remote machine. It’s a good
            practice to set connect timeouts to slightly larger than a multiple
            of 3, which is the default TCP packet retransmission window. Once
            your client has connected to the server and sent the HTTP request,
            the *read* timeout is the number of seconds the client will wait for
            the server to send a response. If you specify a single value for the
            timeout, like this: ``timeout=5``. The timeout value will be applied
            to both the ``connect`` and the ``read`` timeouts. Specify a tuple
            if you would like to set the values separately:
            ``timeout=(3.05, 27)``. If the remote server is very slow, you can
            tell Requests to wait forever for a response, by passing ``None`` as
            a timeout value: ``timeout=None``.

        :param retry_total:
            Total number of retries to allow. Set to ``None`` to remove this
            constraint and fall back on other counts. Set to ``0`` to fail on
            the first retry. Set to ``False`` to disable.

        :param retry_allowed_methods:
            Set of uppercased HTTP method verbs that we should retry on. We try
            on non-idempotent methods to handle real world scenarios. Defaults
            to ``["HEAD", "GET", "POST", "PUT", "DELETE", "OPTIONS", "TRACE"]``.

        :param retry_status_forcelist:
            A set of integer HTTP status codes that we should force a retry on.
            A retry is initiated if the request method is in ``allowed_methods``
            and the response status code is in ``status_forcelist``. Again we
            try on 5XX error codes to handle real world scenarios. Defaults to
            ``[413, 429, 500, 502, 503, 504]``.

        :param retry_backoff_factor:
            A backoff factor to apply between attempts after the second try. The
            delay is calculated as:
            ``{backoff factor} * (2 ** ({number of total retries} - 1))``. If
            the ``backoff_factor`` is ``0.1``, then the delay is
            ``[0.0s, 0.2s, 0.4s, …]``.
        """
        # set some defaults
        if not retry_allowed_methods:
            retry_allowed_methods = ["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]
        if not retry_status_forcelist:
            retry_status_forcelist = [413, 429, 500, 502, 503, 504]

        # instantiate the retry strategy
        self._retries = Retry(
            total=retry_total,
            method_whitelist=retry_allowed_methods,
            status_forcelist=retry_status_forcelist,
            backoff_factor=retry_backoff_factor,
        )

        # instantiate the timeout adapter
        self._timeout_adapter = TimeoutHTTPAdapter(
            timeout=timeout, max_retries=self._retries
        )

        # instantiate the session
        self._session = requests.Session()

    def get_session(self):
        # add the hook to assert the response status
        self._session.hooks["response"] = [assert_status_hook]
        # add the timeout adapter
        self._session.mount("https://", self._timeout_adapter)
        self._session.mount("http://", self._timeout_adapter)
        return self._session