Skip to content

Instantly share code, notes, and snippets.

@layoaster
Created February 4, 2020 13:50
Show Gist options
  • Save layoaster/9fbc02cbd49f7b7f69c938746487426e to your computer and use it in GitHub Desktop.
Save layoaster/9fbc02cbd49f7b7f69c938746487426e to your computer and use it in GitHub Desktop.
Sample of a wrapper to consum a REST API with retry logic
"""
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
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment