Sample of a wrapper to consum a REST API with retry logic

api_request_wrapper.py

"""
This module provides with a high-level Planhat REST API interface. The API is
documented on `Planhat Docs <https://docs.planhat.com/>`_.
"""
import logging
from typing import Any, Dict, List, Union

import backoff
import requests

from config.configuration import PlanhatApiConfiguration
from databases.cache import PlanhatCache
from databases.serializers import CompanyPayload, ContactPayload, TeamMemberResponse

# Type hinting aliases
CustomFields = Dict[str, Any]
ApiCallReturn = Union[str, None]
JsonResponse = Union[List[dict], dict]

logger = logging.getLogger('oncrm')
# Adding our custom logging handlers to the backoff library logger
for handler in logger.handlers:
    logging.getLogger('backoff').addHandler(handler)


class PlanhatApi:
    """
    Wrapper around the raw Planhat REST API. Implements high-level operations
    on Planhat.
    """

    #: The configuration for this class
    planhat_conf = PlanhatApiConfiguration

    def __init__(self, init: bool=False, dry_run: bool=False, force_update: bool=False):
        """
        Initializes the class by constructing a `Session` object.
        :param init: Indicates whether Planhat has no previous config.
            Meaning that some initial configuration (like Planhat's
            `Custom Fields`) has to be done before importing data.
        :param dry_run: If `True`, no Planhat API calls will be made.
        :param force_update: If `True`, all data will be updated on Planhat
            regardless data is outdated or not. This is useful to make
            corrections to data previously uploaded.
        """
        # Setting up requests session
        self.session = requests.Session()
        # session headers
        self.session.headers['Authorization'] = ' '.join(('Bearer', self.planhat_conf.api_key))
        self.session.headers['Content-Type'] = 'application/json'

        # Setting up cache
        self.cache = PlanhatCache()

        self.dry_run = dry_run
        self.force_update = force_update
        if init:
            self._create_custom_fields()

    @backoff.on_exception(
        backoff.expo,
        (requests.exceptions.HTTPError, requests.exceptions.ConnectionError,
         requests.exceptions.Timeout),
        max_tries=PlanhatApiConfiguration.max_tries
    )
    def _make_request(self, verb: str, url: str, payload: CustomFields=None) -> requests.Response:
        """
        Makes the required http request falling back to an exponential
        back-off strategy in case of failures.
        :param verb: The http verb of the request. Valid choices:
            [`GET`,`POST`, `PUT`, `PATCH`, 'DELETE`].
        :param url: The http url of the request.
        :param payload: The body of the request or url parameters
            (on `GET` requests).
        :return: The request response if no Http exception is raised.
        """
        if verb == 'GET':
            return self.session.request(
                verb, url, params=payload, timeout=self.planhat_conf.requests_timeout
            )
        elif verb in ('POST', 'PUT', 'DELETE', 'PATCH'):
            return self.session.request(
                verb, url, json=payload, timeout=self.planhat_conf.requests_timeout
            )