CoreLibs-PyPI-All/src/corelibs/requests_handling/caller.py

"""
requests lib interface
V2 call type
"""

from typing import Any, TypedDict, cast
import requests
from requests import exceptions


class ErrorResponse:
    """
    Error response structure. This is returned if a request could not be completed
    """
    def __init__(
        self,
        code: int,
        message: str,
        action: str,
        url: str,
        exception: exceptions.InvalidSchema | exceptions.ReadTimeout | exceptions.ConnectionError | None = None
    ) -> None:
        self.code = code
        self.message = message
        self.action = action
        self.url = url
        self.exception_name = type(exception).__name__ if exception is not None else None
        self.exception_trace = exception if exception is not None else None


class ProxyConfig(TypedDict):
    """
    Socks proxy settings
    """
    type: str
    host: str
    port: str


class Caller:
    """
    requests lib interface
    """

    def __init__(
        self,
        header: dict[str, str],
        verify: bool = True,
        timeout: int = 20,
        proxy: ProxyConfig | None = None,
        cafile: str | None = None
    ):
        self.headers = header
        self.timeout: int = timeout
        self.cafile = cafile
        self.verify = verify
        self.proxy = cast(dict[str, str], proxy) if proxy is not None else None

    def __timeout(self, timeout: int | None) -> int:
        if timeout is not None and timeout >= 0:
            return timeout
        return self.timeout

    def __call(
        self,
        action: str,
        url: str,
        data: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
        timeout: int | None = None
    ) -> requests.Response | ErrorResponse:
        """
        call wrapper, on error returns None

        Args:
            action (str): _description_
            url (str): _description_
            data (dict | None): _description_. Defaults to None.
            params (dict | None): _description_. Defaults to None.

        Returns:
            requests.Response | None: _description_
        """

        if data is None:
            data = {}
        try:
            if action == "get":
                return requests.get(
                    url,
                    params=params,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy,
                    cert=self.cafile
                )
            if action == "post":
                return requests.post(
                    url,
                    params=params,
                    json=data,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy,
                    cert=self.cafile
                )
            if action == "put":
                return requests.put(
                    url,
                    params=params,
                    json=data,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy,
                    cert=self.cafile
                )
            if action == "patch":
                return requests.patch(
                    url,
                    params=params,
                    json=data,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy,
                    cert=self.cafile
                )
            if action == "delete":
                return requests.delete(
                    url,
                    params=params,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy,
                    cert=self.cafile
                )
            return ErrorResponse(
                100,
                f"Unsupported action '{action}'",
                action,
                url
            )
        except exceptions.InvalidSchema as e:
            return ErrorResponse(
                200,
                f"Invalid URL during '{action}' for {url}",
                action,
                url,
                e
            )
        except exceptions.ReadTimeout as e:
            return ErrorResponse(
                300,
                f"Timeout ({self.timeout}s) during '{action}' for {url}",
                action,
                url,
                e
            )
        except exceptions.ConnectionError as e:
            return ErrorResponse(
                400,
                f"Connection error during '{action}' for {url}",
                action,
                url,
                e
            )

    def get(
        self,
        url: str,
        params: dict[str, Any] | None = None,
        timeout: int | None = None
    ) -> requests.Response | ErrorResponse:
        """
        get data

        Args:
            url (str): _description_
            params (dict | None): _description_

        Returns:
            requests.Response: _description_
        """
        return self.__call('get', url, params=params, timeout=timeout)

    def post(
        self,
        url: str,
        data: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
        timeout: int | None = None
    ) -> requests.Response | ErrorResponse:
        """
        post data

        Args:
            url (str): _description_
            data (dict | None): _description_
            params (dict | None): _description_

        Returns:
            requests.Response | None: _description_
        """
        return self.__call('post', url, data, params, timeout=timeout)

    def put(
        self,
        url: str,
        data: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
        timeout: int | None = None
    ) -> requests.Response | ErrorResponse:
        """_summary_

        Args:
            url (str): _description_
            data (dict | None): _description_
            params (dict | None): _description_

        Returns:
            requests.Response | None: _description_
        """
        return self.__call('put', url, data, params, timeout=timeout)

    def patch(
        self,
        url: str,
        data: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
        timeout: int | None = None
    ) -> requests.Response | ErrorResponse:
        """_summary_

        Args:
            url (str): _description_
            data (dict | None): _description_
            params (dict | None): _description_

        Returns:
            requests.Response | None: _description_
        """
        return self.__call('patch', url, data, params, timeout=timeout)

    def delete(
        self,
        url: str,
        params: dict[str, Any] | None = None,
        timeout: int | None = None
    ) -> requests.Response | ErrorResponse:
        """
        delete

        Args:
            url (str): _description_
            params (dict | None): _description_

        Returns:
            requests.Response | None: _description_
        """
        return self.__call('delete', url, params=params, timeout=timeout)

# __END__