This is a wrapper that can be used to do additional external REST call to some API. Since I have to provide support to legacy code so I added three static functions to support them.
However, the question that comes to my mind is where to actually set default value either in the constructor or it should be in the main function.
is_optional = kwargs.get('is_optional', False)
can also be set directly in constructor. So what is correct way to handle such situations?
class ExternalRestService:
def __init__(self, endpoint, method=sc.REQUEST_GET):
self.endpoint = endpoint
self.method = method
self.timeout = sc.REQUEST_TIMEOUT
@staticmethod
def get(endpoint, request_headers, timeout):
"""
Handle GET External Call
"""
logging.debug("About to make an external *** GET *** to url %s", endpoint)
response = requests.get(url=endpoint, headers=request_headers,
timeout=timeout)
return response
@staticmethod
def post(endpoint, request_headers, json_data, timeout):
"""
Handle POST External Call
"""
logging.debug("About to make an external *** POST *** to url %s", endpoint)
request_headers["Content-type"] = "application/json"
response = requests.post(url=endpoint, headers=request_headers, json_data=json_data,
timeout=timeout)
return response
@staticmethod
def put(endpoint, request_headers, json_data, timeout, fallback=False):
"""
Handle PUT External Call
"""
logging.debug("About to make an external *** PUT *** to url %s", endpoint)
request_headers["Content-type"] = "application/json"
response = requests.put(url=endpoint, headers=request_headers, json_data=json_data,
timeout=timeout)
if fallback and response.status_code == 405:
logging.debug("PUT action failed, applying fallback strategy")
response = requests.put(url=endpoint, headers=request_headers, json_data=json_data,
timeout=timeout)
return response
def invoke(self, **kwargs):
"""
Execute the external call
"""
is_fallback = kwargs.get('is_fallback', False)
is_optional = kwargs.get('is_optional', False)
request_headers = kwargs.get('request_headers', {})
json_data = kwargs.get('json_data', {})
response_list = None
try:
logging.info("About to make an external *** %s *** call", self.method)
if self.method == sc.REQUEST_POST:
response = self.post(self.endpoint, request_headers, json_data, self.timeout)
elif self.method == sc.REQUEST_PUT:
response = self.put(self.endpoint, request_headers, json_data,
self.timeout, fallback=is_fallback)
else:
response = self.get(self.endpoint, request_headers, self.timeout)
response_list, error_list = self.__response_handler(response, is_optional)
except requests.RequestException as err:
error_list = err
return response_list, error_list
@staticmethod
def __response_handler(response, is_optional=False):
"""
Handle API response
"""
status_code = response.status_code
if is_optional and status_code == 404:
return response, None
if not 200 <= status_code < 300:
return None, response
response_list = response.json()
return response_list, None