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

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

from typing import Any
import warnings
import requests
# to hide the verfiy warnings because of the bad SSL settings from Netskope, Akamai, etc
warnings.filterwarnings('ignore', message='Unverified HTTPS request')


class Caller:
    """_summary_"""

    def __init__(
        self,
        header: dict[str, str],
        verify: bool = True,
        timeout: int = 20,
        proxy: dict[str, str] | None = None
    ):
        self.headers = header
        self.timeout: int = timeout
        self.cafile = "/Library/Application Support/Netskope/STAgent/data/nscacert.pem"
        self.verify = verify
        self.proxy = proxy

    def __timeout(self, timeout: int | None) -> int:
        if timeout is not None:
            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 | None:
        """
        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:
            response = None
            if action == "get":
                response = requests.get(
                    url,
                    params=params,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy
                )
            elif action == "post":
                response = requests.post(
                    url,
                    params=params,
                    json=data,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy
                )
            elif action == "put":
                response = requests.put(
                    url,
                    params=params,
                    json=data,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy
                )
            elif action == "patch":
                response = requests.patch(
                    url,
                    params=params,
                    json=data,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy
                )
            elif action == "delete":
                response = requests.delete(
                    url,
                    params=params,
                    headers=self.headers,
                    timeout=self.__timeout(timeout),
                    verify=self.verify,
                    proxies=self.proxy
                )
            return response
        except requests.exceptions.InvalidSchema as e:
            print(f"Invalid URL during '{action}' for {url}:\n\t{e}")
            return None
        except requests.exceptions.ReadTimeout as e:
            print(f"Timeout ({self.timeout}s) during '{action}' for {url}:\n\t{e}")
            return None
        except requests.exceptions.ConnectionError as e:
            print(f"Connection error during '{action}' for {url}:\n\t{e}")
            return None

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

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

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

    def post(
        self, url: str, data: dict[str, Any] | None = None, params: dict[str, Any] | None = None
    ) -> requests.Response | None:
        """
        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)

    def put(
        self, url: str, data: dict[str, Any] | None = None, params: dict[str, Any] | None = None
    ) -> requests.Response | None:
        """_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)

    def patch(
        self, url: str, data: dict[str, Any] | None = None, params: dict[str, Any] | None = None
    ) -> requests.Response | None:
        """_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)

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

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

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

# __END__