timeout_session.py

"""A requests.session with a TimeoutAdapter."""
import requests
from requests.adapters import HTTPAdapter

from config import user_agent

REQUEST_TIMEOUT = 10  # reasonable timeout that all HTTP requests should use


class TimeoutAdapter(HTTPAdapter):
    """An HTTP Adapter that enforces an overridable default timeout on HTTP requests."""

    def __init__(self, timeout: int, *args, **kwargs):
        self.default_timeout = timeout
        if kwargs.get("timeout") is not None:
            del kwargs["timeout"]

        super().__init__(*args, **kwargs)

    def send(self, *args, **kwargs) -> requests.Response:
        """Send, but with a timeout always set."""
        if kwargs["timeout"] is None:
            kwargs["timeout"] = self.default_timeout

        return super().send(*args, **kwargs)


def new_session(
    timeout: int = REQUEST_TIMEOUT, session: requests.Session | None = None
) -> requests.Session:
    """
    Create a new requests.Session and override the default HTTPAdapter with a TimeoutAdapter.

    :param timeout: the timeout to set the TimeoutAdapter to, defaults to REQUEST_TIMEOUT
    :param session: the Session object to attach the Adapter to, defaults to a new session
    :return: The created Session
    """
    if session is None:
        session = requests.Session()
        session.headers['User-Agent'] = user_agent

    adapter = TimeoutAdapter(timeout)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session