Base API Client

Generic HTTP REST API client.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

class BaseAPIClient(metaclass=ABCMeta):
    """
    Generic HTTP REST API client.
    """

    # Configuration - Can be overwritten by implementations if needed
    __version__ = "0.3"
    _accepted_schemes: Set[str] = {"https", "https://"}
    _non_cachable_endpoints: Set[str] = set()
    _auto_refresh_accepted_methods: Set[MethodType] = {"GET", "PUT"}
    _http_statuses_to_retry: Set[HTTPStatus] = set()
    _default_timeout: float = 60.0  # in seconds
    _default_retries: int = 3
    _feature_codes: Dict[str, str] = {}
    _api_version: Optional[str] = None
    _cert_path: Union[str, bool] = True
    _minimum_tls_version: Optional[TLSVersion] = None
    _ssl_ciphers: str = ""

    class KeyNames:
        """
        Can be extended by clients' implementations if needed
        """

        ACCESS_TOKEN = "access_token"
        REFRESH_TOKEN = "refresh_token"
        EXPIRATION_DATE = "expiration_date"
        ID_TOKEN = "id_token"
        CLIENT_ID = "client_id"
        AUTH_RESPONSE = "auth_response"
        AUTHORIZATION_HEADER = "Authorization"
        API_VERSION_HEADER = "api-version"
        FEATURE_CODE_HEADER = "FeatureCode"

    def __init__(self, url: str, **kwargs):
        """
        Client initialization

        Args:
            url: API Url

        Keyword Args:
            proxies (Dict): Dictionary with proxy configuration. Following keys are supported:
                `http_proxy`, `https_proxy`, `no_proxy`
            store_request (bool): If True, all obtained request-response pairs will be stored in memory.
                They might be dumped to a JSON file by calling `.dump_stored_requests` method.
                It is useful to collect the communication for the sake of regression tests.
            cache_file_path (Union[Path, str]): Path to the SQLite database file used as a permanent storage.
                If provided, then all request-response pairs will be cached in the SQLite DB.
                It is a useful feature for development purposes, but it's not recommended for production use
                since currently the cache is never invalidated.
            auto_refresh (bool): If True and `refresh_token` is present, then the client will automatically try
                to refresh the access token when it is about to expire (10 minutes before the deadline).
                In order for that to work `_make_access_token_refresh` method has to be correctly implemented.
                Default: `True`
            api_version (str): Version of the API to use. It is optional parameter. If provided then the version is
                attached to each request as a `api-version` header.
            cert_path (Union[Path, str, bool]): Path to the certificate file or `False` to disable certificate.
            unescape_html (bool): If True, the HTML content will be unescaped before decoding it to JSON.

        """
        self.url: str = self._validate_url(url)
        self._proxies: Dict[str, str] = kwargs.pop("proxies", {})
        self._store_requests: bool = kwargs.pop("store_requests", False)
        self._cache_file_path: Union[Path, str, None] = kwargs.pop("cache_file_path", None)
        self._auto_refresh: bool = kwargs.pop("auto_refresh", True)
        self._api_version: Optional[str] = kwargs.pop("api_version", self._api_version)
        self._cert_path: Union[Path, str, bool, None] = kwargs.pop("cert_path", True)
        self._unescape_html: bool = kwargs.pop("unescape_html", True)

        # Initialize shared elements
        self.auth_data: Dict = {}
        self._memory_storage: List = []
        self._db_cache = LiteCache(path=self._cache_file_path) if self._cache_file_path else None
        self._session = self._initialize_session()

    def __repr__(self):
        return f"{self.__class__.__name__}(url={self.url}, version={self.__version__})"

    @property
    @abstractmethod
    def is_logged(self):
        raise NotImplementedError()

    @abstractmethod
    def authenticate(self, *args, **kwargs) -> bool:
        raise NotImplementedError()

    @property
    def access_token(self) -> Optional[str]:
        """The same as `access_token`. Is here for legacy reasons."""
        access_token = self.auth_data.get(self.KeyNames.ACCESS_TOKEN, None)
        return access_token

    @property
    def auth_token(self) -> Optional[str]:
        """The same as `access_token`. Is here for legacy reasons."""
        _auth_token = self.auth_data.get(self.KeyNames.ACCESS_TOKEN, None)
        return _auth_token

    @property
    def refresh_token(self) -> Optional[str]:
        _auth_token = self.auth_data.get(self.KeyNames.REFRESH_TOKEN, None)
        return _auth_token

    @property
    def client_id(self) -> Optional[str]:
        client_id = self.auth_data.get(self.KeyNames.CLIENT_ID, None)
        return client_id

    @property
    def token_expiration_date(self) -> Optional[datetime.datetime]:
        expiration_date = self.auth_data.get(self.KeyNames.EXPIRATION_DATE, None)
        return expiration_date if isinstance(expiration_date, datetime.datetime) else None

    def _make_access_token_refresh(self, force: bool = False) -> bool:
        """
        This method should be implemented by each API Client that has the automatic access token refresh logic.

        By default, no logic is provided, therefore this method returns False.

        Returns:
            In the BaseAPIClient automatic access token refresh can't be implemented so this always returns False.
        """
        return False

    def _add_authorization_to_request(self, request: requests.Request) -> requests.Request:
        """
        Adds authorization the provided request object.

        The default action is to add Bearer Authorization Token to the headers.

        However, if the client need some different authorization method this method should be overwritten.

        Example:
            For clients that communicate with Azure Function based APIs it should rather add `code` to URL params.

        Args:
            request: Request object to be updated.

        Returns:
            Request object with authorization rules added.
        """

        if self.auth_token is not None:
            request.headers[self.KeyNames.AUTHORIZATION_HEADER] = f"Bearer {self.auth_token}"
        return request

    def logout(self):
        """
        Logs out the client by resetting authentication data.
        """
        self.auth_data = {}

    def dump_stored_requests(self, **kwargs) -> str:
        """
        Dumps all already stored request-response pairs into a JSON object represented as a string.

        Keyword Args:
            kwargs (dict): JSON dumps method arguments.

        Returns:
            JSON object that holds a list or request-response objects as a string.
        """
        if not self._store_requests:
            log.info("Client was not initialized with `store_requests=True` flag.")

        json_list = []
        for response in self._memory_storage:
            _request = {"method": response.request.method, "url": response.request.url, "body": response.request.body}
            _response = {"status_code": response.status_code, "text": response.text}
            json_list.append({"request": _request, "response": _response})

        dumped_requests = json.dumps(json_list, **kwargs)
        return dumped_requests

    def _make_request(
        self,
        method: MethodType,
        url: Optional[str] = None,
        endpoint: Optional[str] = None,
        params: Optional[Dict[str, Any]] = None,
        data: Optional[Any] = None,
        json_data: Optional[JSONType] = None,
        **kwargs,
    ) -> Optional[requests.Response]:
        """
        Prepares and executes an HTTP request call.

        Args:
            method: HTTP method to use.
            endpoint: Endpoint to call. Will be joined with base url to make the final url to call.
            url: Full URL to call. If provided, then the `endpoint` parameter is ignored.
            params: Query parameters that will be added to url. Default: nothing.
            data: Data to be added as request body. It is added in the format 'as given'.
                 In most of the cases with real APIs, one would prefer to use `json_data` argument. Default: nothing.
            json_data: JSON data to be added to the request body. It should be provided as an input
                that can be encoded as a JSON object. Default: nothing.

        Keyword Args:
            extra_headers(Dict[str, str]): Extra headers to be added to the request.
                By default, 'User-Agent', 'Content-Type', and 'Authorization' headers are added.
                But sometimes some extra headers needs to be added as well.
            feature_code (str): Custom Feature Code to be added to headers.
                If not provided, the client will try to find an appropriate feature code
                based on internal `_feature_codes` dictionary.
                If nothing is found nor provided, the no feature code header is added.
            content_type (str): Custom Content-Type header. If not provided, then the 'application/json' is used.
            api_version (str): API version to be added to the URL. If not provided, nothing is added.
            timeout (float): Custom Timeout for this request.
                If not provided, then the `_default_timeout` setting is used.
            auth (Any): Custom authentication method to be used for this request.
                If provided, then the client's authorization method is NOT added to the request.
            add_authorization (bool): If True, then the client's authorization method is added to the request.
                This is ignored if `auth` parameter is provided.
                Default: True

        Returns:
            Optional Response object
        """
        extra_headers: Optional[Dict[str, str]] = kwargs.get("extra_headers")
        feature_code: Optional[str] = kwargs.get("feature_code")
        content_type: Optional[str] = kwargs.get("content_type")
        api_version: Optional[str] = kwargs.get("api_version")
        timeout: float = kwargs.get("timeout", self._default_timeout)
        add_authorization: bool = kwargs.get("add_authorization", True)
        auth = kwargs.get("auth", None)

        # Set up URL
        if url is None:
            if endpoint is None:
                raise ValueError("Either `url` or `endpoint` parameter has to be provided.")
            url = self._join_url(self.url, endpoint)

        if params is None:
            params = {}

        # If the feature code was not provided, try to find it
        if feature_code is None and endpoint is not None:
            feature_code = self._get_feature_code(endpoint)
        # Set up headers
        headers = self._prepare_headers(
            feature_code=feature_code,
            content_type=content_type,
            extra_headers=extra_headers,
            api_version=api_version,
        )
        # Build request object
        req = requests.Request(
            method=method, url=url, headers=headers, params=params, json=json_data, data=data, auth=auth
        )
        # Add client specific authorization to the request
        if add_authorization and auth is None:
            req = self._add_authorization_to_request(request=req)
        # Process the request
        response = self._process_request(request=req, timeout=timeout)

        return response

    def _send_request(
        self, request: requests.Request, timeout: Optional[float] = None, no_of_retries: Optional[int] = None
    ) -> Optional[requests.Response]:
        """
        Sends the provided request and return the response.
        It handles timeouts with linear backoff (each try is one times longer). If it fails for `no_of_retries` times,
        then None is returned.

        Args:
            request: Request object
            timeout: Custom Timeout for this request. If not provided, then `_default_timeout` setting is used.
            no_of_retries: Custom number of retries. If not provided, then `_default_no_of_retries` setting is used.

        Returns:
            Optional response object.
        """

        _timeout = timeout if timeout else self._default_timeout
        no_of_retries = no_of_retries if no_of_retries else self._default_retries
        prepared_request = request.prepare()
        log.debug(f"Requesting {prepared_request.method} {prepared_request.url} (body: {str(prepared_request.body)})")
        for retry in range(1, no_of_retries + 1):
            read_timeout = retry * _timeout
            if retry > 1:
                log.debug(f"Retrying for {retry}. time (out of {no_of_retries}), with timeout {read_timeout}s.")
            try:
                # Linear backoff
                response = self._session.send(
                    request=prepared_request, proxies=self._proxies, timeout=read_timeout, verify=self._cert_path
                )

                if response.status_code in self._http_statuses_to_retry:
                    log.debug(f"Got status={response.status_code}, will retry the request")
                    continue

                return response
            except requests.Timeout:
                log.debug("Request reading failed due to Timeout.")

        log.warning(
            f"Reached retry limit={no_of_retries} for request={request}, no successful response from the server"
        )
        return None

    def _prepare_headers(
        self,
        feature_code: Optional[str] = None,
        content_type: Optional[str] = None,
        api_version: Optional[str] = None,
        extra_headers: Optional[Dict] = None,
    ) -> Dict:
        """
        Prepares headers for a request. Two are added by default:

        * 'User-Agent' - identifies the originator of the request. Client name, version and OS is used to build it.
        * 'Content-Type' - identifies the MIME type of the request body

        Args:
            feature_code: Feature code to be added to headers. Will be added as 'FeatureCode' header. Default: `None`
            content_type: Custom Content type headers. Default: `application/json`
            api_version: API version to be added to the headers. Default: `self._api_version`
            extra_headers: Dictionary with extra headers to be added.

        Returns:
            Dictionary with headers.

        """
        content_type = "application/json" if content_type is None else content_type
        api_version = api_version if api_version is not None else self._api_version
        ua = f"{self.__class__.__name__}/{self.__class__.__version__} ({platform.system()} {platform.release()})"
        headers = {"User-Agent": ua, "Content-Type": content_type}
        if feature_code is not None:
            headers[self.KeyNames.FEATURE_CODE_HEADER] = feature_code
        if api_version:
            headers[self.KeyNames.API_VERSION_HEADER] = api_version
        if extra_headers:
            headers.update(extra_headers)
        return headers

    def _get_feature_code(self, requested_endpoint: str) -> Optional[str]:
        """
        Returns a feature code to be used for a given endpoint value. It looks for the feature code value in the
        `_feature_code` attribute of a client.

        Args:
            requested_endpoint: Requested endpoint

        Returns:
            Feature Code value if endpoint found in the client data. Otherwise, `None` is returned.
        """
        for endpoint, code in self._feature_codes.items():
            if requested_endpoint.startswith(endpoint):
                return code
        return None

    def _validate_url(self, url: str) -> str:
        """
        Validates provided `url`. It checks if the url has the correct format and scheme.
        Allowed schemes should be listed in `allowed_schemes` attribute.

        Args:
            url: URL to be validated

        Returns:
            Valid Url

        Raises:
            Value Error: Empty URL, wrong scheme, or invalid URL.

        """

        if not url:
            raise ValueError("Empty URL provided")
        parsed_url = urlparse(url)
        if parsed_url.scheme not in self._accepted_schemes:
            raise ValueError(f"Wrong scheme, only {self._accepted_schemes} are accepted.")
        if not parsed_url.netloc:
            raise ValueError("Invalid URL provided.")
        return url

    def _initialize_session(self, retries: int = 3, backoff_factor: int = 1) -> requests.Session:
        """
        Helper function that returns requests.Session object that can be used to perform all the requests
        by the client.

        When communicating with single HTTP server, like we do in case of this client,
        using session object over the typical requests.get() etc. methods improves performance.
        Disadvantage of using session is that it comes with no retry policy by default.
        It has to be explicitly added by mounting adapter with desired retry strategy.

        If the client has specified minimum TLS version, then TLSAdapter is used to mount the session.
        Otherwise, the default HTTPAdapter is used.

        Session object details: https://docs.python-requests.org/en/latest/user/advanced/#session-objects
        Retry configuration details: https://findwork.dev/blog/advanced-usage-python-requests-timeouts-retries-hooks/
        Transport Adapters details: https://requests.readthedocs.io/en/latest/user/advanced/#transport-adapters

        Args:
            retries: Number of retries
            backoff_factor: Backoff factor

        Returns:
            Configured Sessions object.

        """
        retry_strategy = Retry(total=retries, backoff_factor=backoff_factor)
        session = requests.Session()
        adapter: HTTPAdapter
        if self._minimum_tls_version:
            adapter = TLSAdapter(
                ssl_minimum_version=self._minimum_tls_version,
                ssl_ciphers=self._ssl_ciphers,
                max_retries=retry_strategy,
            )
        else:
            adapter = HTTPAdapter(max_retries=retry_strategy)

        for scheme in self._accepted_schemes:
            session.mount(scheme, adapter)
        return session

    def _process_request(self, request: requests.Request, timeout: float) -> Optional[requests.Response]:
        """
        Produces a response based on the provided request.
        If DB caching is enabled and conditions are met then the response is read from cache.
        Otherwise, the request is sent and the response is received from web.
        Finally, if the memory based response storage is enabled, response object is stored there.

        Args:
            request: Request object.
            timeout: Timeout in seconds.

        Returns:
            Response object if request was successful. Otherwise, `None` is returned.
        """
        response = None
        is_valid_response = False
        is_cacheable = all((endpoint not in request.url for endpoint in self._non_cachable_endpoints))

        # Try to get response from cache
        if self._db_cache and is_cacheable:
            response = self._db_cache[request]

        # If nothing found in cache, we have to send the request
        if response is None:
            # But before that we will check if we have to refresh our access token
            if (
                self._auto_refresh
                and request.method in self._auto_refresh_accepted_methods
                and self._make_access_token_refresh()
            ):
                # Update request with new access token
                request = self._add_authorization_to_request(request)
            # Finally, we are sending the request
            response = self._send_request(request=request, timeout=timeout)
            # Special case added for the access token refresh,
            # if the response is 401 then we will try to force refresh the token and send the request again
            is_not_authenticated_response = isinstance(response, requests.Response) and response.status_code == 401
            if self._auto_refresh and is_not_authenticated_response and self._make_access_token_refresh(force=True):
                request = self._add_authorization_to_request(request)
                response = self._send_request(request=request, timeout=timeout)
            is_valid_response = isinstance(response, requests.Response) and 200 <= response.status_code < 400
        # Store the response, if storing is enabled and response is valid
        # DB cache
        if self._db_cache and is_cacheable and is_valid_response:
            self._db_cache[request] = response
        # Memory storage
        if self._store_requests and is_valid_response:
            self._memory_storage.append(response)
        return response

    @staticmethod
    def _get_status_code(response: Optional[requests.Response]) -> Optional[HTTPStatus]:
        """
        Gets the status code from the response object

        Args:
            response: Response object

        Returns:
            HTTPStatus enum with response's status code. If not possible to get it `None` is returned.
        """

        if not isinstance(response, requests.Response):
            return None

        try:
            return HTTPStatus(response.status_code)
        except ValueError:
            return None

    def _decode_response(self, response: Optional[requests.Response], default: T) -> T:
        """

        Decodes the response object into JSON object. That means, it is assumed that the content of the response
        is encoded in JSON format.

        `default` argument is mandatory, and it is returned when there is any problem with the response
        (error status code, error while decoding).

        Args:
            response: Response object
            default: Default value that should be returned when response is not valid
                or the response's status code indicates error. If `check_type` parameter is set to True,
                the default is also used to determine the expected type of the response.
                e.g. if default is a List, then the response is also expected to be a List.

        Returns:
            JSON decoded response if the response object was valid and of correct type.
                Otherwise, `default` is returned.
        """
        if not isinstance(response, requests.Response):
            return default

        try:
            # Check response status, raise exception on error
            response.raise_for_status()
            # Unescape HTML if specified
            response_text = html.unescape(response.text) if self._unescape_html else response.text
            # Parse response content as JSON
            parsed_response = json.loads(response_text)
            # Check the response type
            if type(parsed_response) is not type(default):
                raise TypeError(
                    "Invalid type of the response. " f"Expected: {type(default)}, received: {type(parsed_response)}"
                )
            return parsed_response
        except (TypeError, requests.exceptions.HTTPError) as exp:
            log.error(exp)
            if response.text:
                log.info(f"API Response: {response.text}")
            log.debug("Traceback:", exc_info=True)
            return default
        except json.JSONDecodeError:
            log.error(f"Error while decoding response content: ({response.text})")
            log.debug("Traceback:", exc_info=True)
            return default

    @staticmethod
    def _join_url(base_url: str, extra_path: str) -> str:
        """
        Joins base url with the extra path with respect to the '/' characters.
        """

        base_url = base_url[:-1] if base_url.endswith("/") else base_url
        extra_path = extra_path[1:] if extra_path.startswith("/") else extra_path
        extra_path = extra_path[:-1] if extra_path.endswith("/") else extra_path

        url = f"{base_url}/{extra_path}"
        return url

    def _decode_response_to_model(
        self, response: Optional[requests.Response], model_type: Type[AcceptedModelType]
    ) -> AcceptedModelType:
        """
        Decodes the received response to the given model type instance.

        First we check if the provided response is of type requests.Response. If not, we raise a TypeError.
        Then we check the response status code, if it is not 2xx, we raise an HTTPError exception.
        Next, we parse the response content as JSON and validate it against the provided model type.
        If that fails, we raise an exception ValidationError exceptions.

        Finally, we return the model instance with the decoded response content.

        Args:
            response: Response object received from the API.
            model_type: Model class that the response content will be decoded to.

        Returns:
            Instance of the model class with the decoded response content.

        Raises:
            TypeError: If the response is not a requests.Response object
            HTTPError: If the response status code is not >=400
            ValidationError: If the response content cannot be parsed as JSON or does not match the expected model

        """
        if not isinstance(response, requests.Response):
            raise TypeError(f"Invalid response type: {type(response)}")

        try:
            # Check response status, raise exception on error
            response.raise_for_status()
            # Unescape HTML if specified
            response_text = html.unescape(response.text) if self._unescape_html else response.text
            # Initialize the specific model class based on the model type
            # mypy complains about the following line, but it is correct
            response_model = model_type.model_validate_json(response_text)
            return response_model

        except requests.exceptions.HTTPError as http_error:
            log.error(http_error)
            if response.text:
                log.info(f"API Response: {response.text}")
            log.debug("Traceback: ", exc_info=True)
            raise http_error
        except ValidationError as validation_error:
            log.error(validation_error)
            log.debug("Traceback: ", exc_info=True)
            raise validation_error

KeyNames

Can be extended by clients' implementations if needed

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

class KeyNames:
    """
    Can be extended by clients' implementations if needed
    """

    ACCESS_TOKEN = "access_token"
    REFRESH_TOKEN = "refresh_token"
    EXPIRATION_DATE = "expiration_date"
    ID_TOKEN = "id_token"
    CLIENT_ID = "client_id"
    AUTH_RESPONSE = "auth_response"
    AUTHORIZATION_HEADER = "Authorization"
    API_VERSION_HEADER = "api-version"
    FEATURE_CODE_HEADER = "FeatureCode"

__init__(url, **kwargs)

Client initialization

Parameters:

Name	Type	Description	Default
url	str	API Url	required

Other Parameters:

Name	Type	Description
proxies	Dict	Dictionary with proxy configuration. Following keys are supported: http_proxy, https_proxy, no_proxy
store_request	bool	If True, all obtained request-response pairs will be stored in memory. They might be dumped to a JSON file by calling .dump_stored_requests method. It is useful to collect the communication for the sake of regression tests.
cache_file_path	Union[Path, str]	Path to the SQLite database file used as a permanent storage. If provided, then all request-response pairs will be cached in the SQLite DB. It is a useful feature for development purposes, but it's not recommended for production use since currently the cache is never invalidated.
auto_refresh	bool	If True and refresh_token is present, then the client will automatically try to refresh the access token when it is about to expire (10 minutes before the deadline). In order for that to work _make_access_token_refresh method has to be correctly implemented. Default: True
api_version	str	Version of the API to use. It is optional parameter. If provided then the version is attached to each request as a api-version header.
cert_path	Union[Path, str, bool]	Path to the certificate file or False to disable certificate.
unescape_html	bool	If True, the HTML content will be unescaped before decoding it to JSON.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def __init__(self, url: str, **kwargs):
    """
    Client initialization

    Args:
        url: API Url

    Keyword Args:
        proxies (Dict): Dictionary with proxy configuration. Following keys are supported:
            `http_proxy`, `https_proxy`, `no_proxy`
        store_request (bool): If True, all obtained request-response pairs will be stored in memory.
            They might be dumped to a JSON file by calling `.dump_stored_requests` method.
            It is useful to collect the communication for the sake of regression tests.
        cache_file_path (Union[Path, str]): Path to the SQLite database file used as a permanent storage.
            If provided, then all request-response pairs will be cached in the SQLite DB.
            It is a useful feature for development purposes, but it's not recommended for production use
            since currently the cache is never invalidated.
        auto_refresh (bool): If True and `refresh_token` is present, then the client will automatically try
            to refresh the access token when it is about to expire (10 minutes before the deadline).
            In order for that to work `_make_access_token_refresh` method has to be correctly implemented.
            Default: `True`
        api_version (str): Version of the API to use. It is optional parameter. If provided then the version is
            attached to each request as a `api-version` header.
        cert_path (Union[Path, str, bool]): Path to the certificate file or `False` to disable certificate.
        unescape_html (bool): If True, the HTML content will be unescaped before decoding it to JSON.

    """
    self.url: str = self._validate_url(url)
    self._proxies: Dict[str, str] = kwargs.pop("proxies", {})
    self._store_requests: bool = kwargs.pop("store_requests", False)
    self._cache_file_path: Union[Path, str, None] = kwargs.pop("cache_file_path", None)
    self._auto_refresh: bool = kwargs.pop("auto_refresh", True)
    self._api_version: Optional[str] = kwargs.pop("api_version", self._api_version)
    self._cert_path: Union[Path, str, bool, None] = kwargs.pop("cert_path", True)
    self._unescape_html: bool = kwargs.pop("unescape_html", True)

    # Initialize shared elements
    self.auth_data: Dict = {}
    self._memory_storage: List = []
    self._db_cache = LiteCache(path=self._cache_file_path) if self._cache_file_path else None
    self._session = self._initialize_session()

authenticate(*args, **kwargs) abstractmethod

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

@abstractmethod
def authenticate(self, *args, **kwargs) -> bool:
    raise NotImplementedError()

logout()

Logs out the client by resetting authentication data.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def logout(self):
    """
    Logs out the client by resetting authentication data.
    """
    self.auth_data = {}

dump_stored_requests(**kwargs)

Dumps all already stored request-response pairs into a JSON object represented as a string.

Other Parameters:

Name	Type	Description
kwargs	dict	JSON dumps method arguments.

Returns:

Type	Description
str	JSON object that holds a list or request-response objects as a string.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def dump_stored_requests(self, **kwargs) -> str:
    """
    Dumps all already stored request-response pairs into a JSON object represented as a string.

    Keyword Args:
        kwargs (dict): JSON dumps method arguments.

    Returns:
        JSON object that holds a list or request-response objects as a string.
    """
    if not self._store_requests:
        log.info("Client was not initialized with `store_requests=True` flag.")

    json_list = []
    for response in self._memory_storage:
        _request = {"method": response.request.method, "url": response.request.url, "body": response.request.body}
        _response = {"status_code": response.status_code, "text": response.text}
        json_list.append({"request": _request, "response": _response})

    dumped_requests = json.dumps(json_list, **kwargs)
    return dumped_requests

_validate_url(url)

Validates provided url. It checks if the url has the correct format and scheme. Allowed schemes should be listed in allowed_schemes attribute.

Parameters:

Name	Type	Description	Default
url	str	URL to be validated	required

Returns:

Type	Description
str	Valid Url

Raises:

Type	Description
Value Error	Empty URL, wrong scheme, or invalid URL.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _validate_url(self, url: str) -> str:
    """
    Validates provided `url`. It checks if the url has the correct format and scheme.
    Allowed schemes should be listed in `allowed_schemes` attribute.

    Args:
        url: URL to be validated

    Returns:
        Valid Url

    Raises:
        Value Error: Empty URL, wrong scheme, or invalid URL.

    """

    if not url:
        raise ValueError("Empty URL provided")
    parsed_url = urlparse(url)
    if parsed_url.scheme not in self._accepted_schemes:
        raise ValueError(f"Wrong scheme, only {self._accepted_schemes} are accepted.")
    if not parsed_url.netloc:
        raise ValueError("Invalid URL provided.")
    return url

_initialize_session(retries=3, backoff_factor=1)

Helper function that returns requests.Session object that can be used to perform all the requests by the client.

When communicating with single HTTP server, like we do in case of this client, using session object over the typical requests.get() etc. methods improves performance. Disadvantage of using session is that it comes with no retry policy by default. It has to be explicitly added by mounting adapter with desired retry strategy.

If the client has specified minimum TLS version, then TLSAdapter is used to mount the session. Otherwise, the default HTTPAdapter is used.

Session object details: https://docs.python-requests.org/en/latest/user/advanced/#session-objects Retry configuration details: https://findwork.dev/blog/advanced-usage-python-requests-timeouts-retries-hooks/ Transport Adapters details: https://requests.readthedocs.io/en/latest/user/advanced/#transport-adapters

Parameters:

Name	Type	Description	Default
retries	int	Number of retries	3
backoff_factor	int	Backoff factor	1

Returns:

Type	Description
Session	Configured Sessions object.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _initialize_session(self, retries: int = 3, backoff_factor: int = 1) -> requests.Session:
    """
    Helper function that returns requests.Session object that can be used to perform all the requests
    by the client.

    When communicating with single HTTP server, like we do in case of this client,
    using session object over the typical requests.get() etc. methods improves performance.
    Disadvantage of using session is that it comes with no retry policy by default.
    It has to be explicitly added by mounting adapter with desired retry strategy.

    If the client has specified minimum TLS version, then TLSAdapter is used to mount the session.
    Otherwise, the default HTTPAdapter is used.

    Session object details: https://docs.python-requests.org/en/latest/user/advanced/#session-objects
    Retry configuration details: https://findwork.dev/blog/advanced-usage-python-requests-timeouts-retries-hooks/
    Transport Adapters details: https://requests.readthedocs.io/en/latest/user/advanced/#transport-adapters

    Args:
        retries: Number of retries
        backoff_factor: Backoff factor

    Returns:
        Configured Sessions object.

    """
    retry_strategy = Retry(total=retries, backoff_factor=backoff_factor)
    session = requests.Session()
    adapter: HTTPAdapter
    if self._minimum_tls_version:
        adapter = TLSAdapter(
            ssl_minimum_version=self._minimum_tls_version,
            ssl_ciphers=self._ssl_ciphers,
            max_retries=retry_strategy,
        )
    else:
        adapter = HTTPAdapter(max_retries=retry_strategy)

    for scheme in self._accepted_schemes:
        session.mount(scheme, adapter)
    return session

_make_access_token_refresh(force=False)

This method should be implemented by each API Client that has the automatic access token refresh logic.

By default, no logic is provided, therefore this method returns False.

Returns:

Type	Description
bool	In the BaseAPIClient automatic access token refresh can't be implemented so this always returns False.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _make_access_token_refresh(self, force: bool = False) -> bool:
    """
    This method should be implemented by each API Client that has the automatic access token refresh logic.

    By default, no logic is provided, therefore this method returns False.

    Returns:
        In the BaseAPIClient automatic access token refresh can't be implemented so this always returns False.
    """
    return False

_make_request(method, url=None, endpoint=None, params=None, data=None, json_data=None, **kwargs)

Prepares and executes an HTTP request call.

Parameters:

Name	Type	Description	Default
method	MethodType	HTTP method to use.	required
endpoint	Optional[str]	Endpoint to call. Will be joined with base url to make the final url to call.	None
url	Optional[str]	Full URL to call. If provided, then the endpoint parameter is ignored.	None
params	Optional[Dict[str, Any]]	Query parameters that will be added to url. Default: nothing.	None
data	Optional[Any]	Data to be added as request body. It is added in the format 'as given'. In most of the cases with real APIs, one would prefer to use json_data argument. Default: nothing.	None
json_data	Optional[JSONType]	JSON data to be added to the request body. It should be provided as an input that can be encoded as a JSON object. Default: nothing.	None

Other Parameters:

Name	Type	Description
extra_headers	Dict[str, str]	Extra headers to be added to the request. By default, 'User-Agent', 'Content-Type', and 'Authorization' headers are added. But sometimes some extra headers needs to be added as well.
feature_code	str	Custom Feature Code to be added to headers. If not provided, the client will try to find an appropriate feature code based on internal _feature_codes dictionary. If nothing is found nor provided, the no feature code header is added.
content_type	str	Custom Content-Type header. If not provided, then the 'application/json' is used.
api_version	str	API version to be added to the URL. If not provided, nothing is added.
timeout	float	Custom Timeout for this request. If not provided, then the _default_timeout setting is used.
auth	Any	Custom authentication method to be used for this request. If provided, then the client's authorization method is NOT added to the request.
add_authorization	bool	If True, then the client's authorization method is added to the request. This is ignored if auth parameter is provided. Default: True

Returns:

Type	Description
Optional[Response]	Optional Response object

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _make_request(
    self,
    method: MethodType,
    url: Optional[str] = None,
    endpoint: Optional[str] = None,
    params: Optional[Dict[str, Any]] = None,
    data: Optional[Any] = None,
    json_data: Optional[JSONType] = None,
    **kwargs,
) -> Optional[requests.Response]:
    """
    Prepares and executes an HTTP request call.

    Args:
        method: HTTP method to use.
        endpoint: Endpoint to call. Will be joined with base url to make the final url to call.
        url: Full URL to call. If provided, then the `endpoint` parameter is ignored.
        params: Query parameters that will be added to url. Default: nothing.
        data: Data to be added as request body. It is added in the format 'as given'.
             In most of the cases with real APIs, one would prefer to use `json_data` argument. Default: nothing.
        json_data: JSON data to be added to the request body. It should be provided as an input
            that can be encoded as a JSON object. Default: nothing.

    Keyword Args:
        extra_headers(Dict[str, str]): Extra headers to be added to the request.
            By default, 'User-Agent', 'Content-Type', and 'Authorization' headers are added.
            But sometimes some extra headers needs to be added as well.
        feature_code (str): Custom Feature Code to be added to headers.
            If not provided, the client will try to find an appropriate feature code
            based on internal `_feature_codes` dictionary.
            If nothing is found nor provided, the no feature code header is added.
        content_type (str): Custom Content-Type header. If not provided, then the 'application/json' is used.
        api_version (str): API version to be added to the URL. If not provided, nothing is added.
        timeout (float): Custom Timeout for this request.
            If not provided, then the `_default_timeout` setting is used.
        auth (Any): Custom authentication method to be used for this request.
            If provided, then the client's authorization method is NOT added to the request.
        add_authorization (bool): If True, then the client's authorization method is added to the request.
            This is ignored if `auth` parameter is provided.
            Default: True

    Returns:
        Optional Response object
    """
    extra_headers: Optional[Dict[str, str]] = kwargs.get("extra_headers")
    feature_code: Optional[str] = kwargs.get("feature_code")
    content_type: Optional[str] = kwargs.get("content_type")
    api_version: Optional[str] = kwargs.get("api_version")
    timeout: float = kwargs.get("timeout", self._default_timeout)
    add_authorization: bool = kwargs.get("add_authorization", True)
    auth = kwargs.get("auth", None)

    # Set up URL
    if url is None:
        if endpoint is None:
            raise ValueError("Either `url` or `endpoint` parameter has to be provided.")
        url = self._join_url(self.url, endpoint)

    if params is None:
        params = {}

    # If the feature code was not provided, try to find it
    if feature_code is None and endpoint is not None:
        feature_code = self._get_feature_code(endpoint)
    # Set up headers
    headers = self._prepare_headers(
        feature_code=feature_code,
        content_type=content_type,
        extra_headers=extra_headers,
        api_version=api_version,
    )
    # Build request object
    req = requests.Request(
        method=method, url=url, headers=headers, params=params, json=json_data, data=data, auth=auth
    )
    # Add client specific authorization to the request
    if add_authorization and auth is None:
        req = self._add_authorization_to_request(request=req)
    # Process the request
    response = self._process_request(request=req, timeout=timeout)

    return response

_prepare_headers(feature_code=None, content_type=None, api_version=None, extra_headers=None)

Prepares headers for a request. Two are added by default:

• 'User-Agent' - identifies the originator of the request. Client name, version and OS is used to build it.
• 'Content-Type' - identifies the MIME type of the request body

Parameters:

Name	Type	Description	Default
feature_code	Optional[str]	Feature code to be added to headers. Will be added as 'FeatureCode' header. Default: None	None
content_type	Optional[str]	Custom Content type headers. Default: application/json	None
api_version	Optional[str]	API version to be added to the headers. Default: self._api_version	None
extra_headers	Optional[Dict]	Dictionary with extra headers to be added.	None

Returns:

Type	Description
Dict	Dictionary with headers.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _prepare_headers(
    self,
    feature_code: Optional[str] = None,
    content_type: Optional[str] = None,
    api_version: Optional[str] = None,
    extra_headers: Optional[Dict] = None,
) -> Dict:
    """
    Prepares headers for a request. Two are added by default:

    * 'User-Agent' - identifies the originator of the request. Client name, version and OS is used to build it.
    * 'Content-Type' - identifies the MIME type of the request body

    Args:
        feature_code: Feature code to be added to headers. Will be added as 'FeatureCode' header. Default: `None`
        content_type: Custom Content type headers. Default: `application/json`
        api_version: API version to be added to the headers. Default: `self._api_version`
        extra_headers: Dictionary with extra headers to be added.

    Returns:
        Dictionary with headers.

    """
    content_type = "application/json" if content_type is None else content_type
    api_version = api_version if api_version is not None else self._api_version
    ua = f"{self.__class__.__name__}/{self.__class__.__version__} ({platform.system()} {platform.release()})"
    headers = {"User-Agent": ua, "Content-Type": content_type}
    if feature_code is not None:
        headers[self.KeyNames.FEATURE_CODE_HEADER] = feature_code
    if api_version:
        headers[self.KeyNames.API_VERSION_HEADER] = api_version
    if extra_headers:
        headers.update(extra_headers)
    return headers

_get_feature_code(requested_endpoint)

Returns a feature code to be used for a given endpoint value. It looks for the feature code value in the _feature_code attribute of a client.

Parameters:

Name	Type	Description	Default
requested_endpoint	str	Requested endpoint	required

Returns:

Type	Description
Optional[str]	Feature Code value if endpoint found in the client data. Otherwise, None is returned.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _get_feature_code(self, requested_endpoint: str) -> Optional[str]:
    """
    Returns a feature code to be used for a given endpoint value. It looks for the feature code value in the
    `_feature_code` attribute of a client.

    Args:
        requested_endpoint: Requested endpoint

    Returns:
        Feature Code value if endpoint found in the client data. Otherwise, `None` is returned.
    """
    for endpoint, code in self._feature_codes.items():
        if requested_endpoint.startswith(endpoint):
            return code
    return None

_add_authorization_to_request(request)

Adds authorization the provided request object.

The default action is to add Bearer Authorization Token to the headers.

However, if the client need some different authorization method this method should be overwritten.

Example

For clients that communicate with Azure Function based APIs it should rather add code to URL params.

Parameters:

Name	Type	Description	Default
request	Request	Request object to be updated.	required

Returns:

Type	Description
Request	Request object with authorization rules added.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _add_authorization_to_request(self, request: requests.Request) -> requests.Request:
    """
    Adds authorization the provided request object.

    The default action is to add Bearer Authorization Token to the headers.

    However, if the client need some different authorization method this method should be overwritten.

    Example:
        For clients that communicate with Azure Function based APIs it should rather add `code` to URL params.

    Args:
        request: Request object to be updated.

    Returns:
        Request object with authorization rules added.
    """

    if self.auth_token is not None:
        request.headers[self.KeyNames.AUTHORIZATION_HEADER] = f"Bearer {self.auth_token}"
    return request

_process_request(request, timeout)

Produces a response based on the provided request. If DB caching is enabled and conditions are met then the response is read from cache. Otherwise, the request is sent and the response is received from web. Finally, if the memory based response storage is enabled, response object is stored there.

Parameters:

Name	Type	Description	Default
request	Request	Request object.	required
timeout	float	Timeout in seconds.	required

Returns:

Type	Description
Optional[Response]	Response object if request was successful. Otherwise, None is returned.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _process_request(self, request: requests.Request, timeout: float) -> Optional[requests.Response]:
    """
    Produces a response based on the provided request.
    If DB caching is enabled and conditions are met then the response is read from cache.
    Otherwise, the request is sent and the response is received from web.
    Finally, if the memory based response storage is enabled, response object is stored there.

    Args:
        request: Request object.
        timeout: Timeout in seconds.

    Returns:
        Response object if request was successful. Otherwise, `None` is returned.
    """
    response = None
    is_valid_response = False
    is_cacheable = all((endpoint not in request.url for endpoint in self._non_cachable_endpoints))

    # Try to get response from cache
    if self._db_cache and is_cacheable:
        response = self._db_cache[request]

    # If nothing found in cache, we have to send the request
    if response is None:
        # But before that we will check if we have to refresh our access token
        if (
            self._auto_refresh
            and request.method in self._auto_refresh_accepted_methods
            and self._make_access_token_refresh()
        ):
            # Update request with new access token
            request = self._add_authorization_to_request(request)
        # Finally, we are sending the request
        response = self._send_request(request=request, timeout=timeout)
        # Special case added for the access token refresh,
        # if the response is 401 then we will try to force refresh the token and send the request again
        is_not_authenticated_response = isinstance(response, requests.Response) and response.status_code == 401
        if self._auto_refresh and is_not_authenticated_response and self._make_access_token_refresh(force=True):
            request = self._add_authorization_to_request(request)
            response = self._send_request(request=request, timeout=timeout)
        is_valid_response = isinstance(response, requests.Response) and 200 <= response.status_code < 400
    # Store the response, if storing is enabled and response is valid
    # DB cache
    if self._db_cache and is_cacheable and is_valid_response:
        self._db_cache[request] = response
    # Memory storage
    if self._store_requests and is_valid_response:
        self._memory_storage.append(response)
    return response

_send_request(request, timeout=None, no_of_retries=None)

Sends the provided request and return the response. It handles timeouts with linear backoff (each try is one times longer). If it fails for no_of_retries times, then None is returned.

Parameters:

Name	Type	Description	Default
request	Request	Request object	required
timeout	Optional[float]	Custom Timeout for this request. If not provided, then _default_timeout setting is used.	None
no_of_retries	Optional[int]	Custom number of retries. If not provided, then _default_no_of_retries setting is used.	None

Returns:

Type	Description
Optional[Response]	Optional response object.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _send_request(
    self, request: requests.Request, timeout: Optional[float] = None, no_of_retries: Optional[int] = None
) -> Optional[requests.Response]:
    """
    Sends the provided request and return the response.
    It handles timeouts with linear backoff (each try is one times longer). If it fails for `no_of_retries` times,
    then None is returned.

    Args:
        request: Request object
        timeout: Custom Timeout for this request. If not provided, then `_default_timeout` setting is used.
        no_of_retries: Custom number of retries. If not provided, then `_default_no_of_retries` setting is used.

    Returns:
        Optional response object.
    """

    _timeout = timeout if timeout else self._default_timeout
    no_of_retries = no_of_retries if no_of_retries else self._default_retries
    prepared_request = request.prepare()
    log.debug(f"Requesting {prepared_request.method} {prepared_request.url} (body: {str(prepared_request.body)})")
    for retry in range(1, no_of_retries + 1):
        read_timeout = retry * _timeout
        if retry > 1:
            log.debug(f"Retrying for {retry}. time (out of {no_of_retries}), with timeout {read_timeout}s.")
        try:
            # Linear backoff
            response = self._session.send(
                request=prepared_request, proxies=self._proxies, timeout=read_timeout, verify=self._cert_path
            )

            if response.status_code in self._http_statuses_to_retry:
                log.debug(f"Got status={response.status_code}, will retry the request")
                continue

            return response
        except requests.Timeout:
            log.debug("Request reading failed due to Timeout.")

    log.warning(
        f"Reached retry limit={no_of_retries} for request={request}, no successful response from the server"
    )
    return None

_decode_response(response, default)

Decodes the response object into JSON object. That means, it is assumed that the content of the response is encoded in JSON format.

default argument is mandatory, and it is returned when there is any problem with the response (error status code, error while decoding).

Parameters:

Name	Type	Description	Default
response	Optional[Response]	Response object	required
default	T	Default value that should be returned when response is not valid or the response's status code indicates error. If check_type parameter is set to True, the default is also used to determine the expected type of the response. e.g. if default is a List, then the response is also expected to be a List.	required

Returns:

Type	Description
T	JSON decoded response if the response object was valid and of correct type. Otherwise, default is returned.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

def _decode_response(self, response: Optional[requests.Response], default: T) -> T:
    """

    Decodes the response object into JSON object. That means, it is assumed that the content of the response
    is encoded in JSON format.

    `default` argument is mandatory, and it is returned when there is any problem with the response
    (error status code, error while decoding).

    Args:
        response: Response object
        default: Default value that should be returned when response is not valid
            or the response's status code indicates error. If `check_type` parameter is set to True,
            the default is also used to determine the expected type of the response.
            e.g. if default is a List, then the response is also expected to be a List.

    Returns:
        JSON decoded response if the response object was valid and of correct type.
            Otherwise, `default` is returned.
    """
    if not isinstance(response, requests.Response):
        return default

    try:
        # Check response status, raise exception on error
        response.raise_for_status()
        # Unescape HTML if specified
        response_text = html.unescape(response.text) if self._unescape_html else response.text
        # Parse response content as JSON
        parsed_response = json.loads(response_text)
        # Check the response type
        if type(parsed_response) is not type(default):
            raise TypeError(
                "Invalid type of the response. " f"Expected: {type(default)}, received: {type(parsed_response)}"
            )
        return parsed_response
    except (TypeError, requests.exceptions.HTTPError) as exp:
        log.error(exp)
        if response.text:
            log.info(f"API Response: {response.text}")
        log.debug("Traceback:", exc_info=True)
        return default
    except json.JSONDecodeError:
        log.error(f"Error while decoding response content: ({response.text})")
        log.debug("Traceback:", exc_info=True)
        return default

_get_status_code(response) staticmethod

Gets the status code from the response object

Parameters:

Name	Type	Description	Default
response	Optional[Response]	Response object	required

Returns:

Type	Description
Optional[HTTPStatus]	HTTPStatus enum with response's status code. If not possible to get it None is returned.

Source code in reportconnectors/api_client/base_api_client/base_api_client.py

@staticmethod
def _get_status_code(response: Optional[requests.Response]) -> Optional[HTTPStatus]:
    """
    Gets the status code from the response object

    Args:
        response: Response object

    Returns:
        HTTPStatus enum with response's status code. If not possible to get it `None` is returned.
    """

    if not isinstance(response, requests.Response):
        return None

    try:
        return HTTPStatus(response.status_code)
    except ValueError:
        return None