Skip to content

Instantly share code, notes, and snippets.

@HacKanCuBa HacKanCuBa/connector.py
Last active Jun 7, 2019

Embed
What would you like to do?
"""Wrappers around Python 3 Requests library.
This lib will log errors, warnings and request duration, not raising any
exception: in such error cases, an empty dict is returned. To identify, if
necessary, that there where errors, a with_error flag must be set in the
arguments so that the methods return a tuple in the form of
(response_data: any, error: bool).
If there's any response expected from the endpoint, it will be returned
JSON-converted as-is, which means it's either valid JSON (string, number,
list, dict) or an empty dict (default response value, which is still valid
JSON).
Examples:
response = JSONConnector.get('https://...', timeout=5)
other_response, error = JSONConnector('https://...', timeout=3, with_error=True)
send_headers = JSONConnector.request('HEAD', 'https://...', timeout=10,
headers={'X-Auth': 'ab12...023'})
"""
import logging
from time import time
from typing import Tuple, Union
import requests
from django.conf import settings
__version__ = '0.2.2'
__author__ = 'HacKan (https://hackan.net)'
__license__ = 'GPL-3+'
logger = logging.getLogger(__name__)
class JSONConnector:
"""Generic independent class to handle JSON endpoints."""
VERIFY_SSL = getattr(settings, 'REQUESTS_VERIFY_SSL', True)
@staticmethod
def request(method: str, url: str, *, timeout: Union[int, float],
with_error: bool = False, **kwargs) -> Union[any, Tuple[any, bool]]:
"""Make a request to a JSON endpoint, return the JSON converted response if any.
Request time, errors and exceptions are all logged using the standard
logger.
Note that the type of the returned response depends on the endpoint,
but it will always be some valid JSON or an empty dict.
To know whether an error occurred or not, set with_error to True: it
will then return a tuple in the form of (json_response: any, error: bool).
method: The request method as string, such as GET, POST, PUT, etc.
url: Endpoint URL as string.
timeout: Connection timeout in seconds (0 for inf).
with_error: Return a tuple in the form (json_response: any, error: bool).
"""
error = False
if timeout > 0:
kwargs['timeout'] = timeout
if 'verify' not in kwargs:
kwargs['verify'] = JSONConnector.VERIFY_SSL
if 'headers' not in kwargs:
kwargs['headers'] = {
'content-type': 'application/json',
}
response_data = {}
request_time_start = time()
try:
response = requests.request(method, url, **kwargs)
request_time_end = time()
except (requests.exceptions.ConnectionError,
requests.exceptions.ReadTimeout) as e:
request_time_end = time()
logger.error(
'Error [%s]ing data (kwargs: %s) to/from the endpoint (url: %s): %s',
method,
str(kwargs),
url,
str(e)
)
error = True
else:
if response.ok:
try:
response_data = response.json()
except ValueError:
logger.warning(
'Response from endpoint %s is not valid JSON: %d %s',
url,
response.status_code,
response.text
)
else:
logger.warning(
'Response from endpoint %s (kwargs: %s) is NOT OK: %d %s',
url,
str(kwargs),
response.status_code,
response.text
)
error = True
logger.debug(
'Request to endpoint %s took %s seconds',
url,
'{:.2f}'.format(request_time_end - request_time_start)
)
return (response_data, error) if with_error else response_data
@staticmethod
def get(url: str, *, timeout: Union[int, float], with_error: bool = False,
**kwargs) -> Union[any, Tuple[any, bool]]:
"""Retrieve data from a JSON endpoint, return the JSON converted response if any.
Request time, errors and exceptions are all logged using the standard
logger.
Note that the type of the returned response depends on the endpoint,
but it will always be some valid JSON or an empty dict.
To know whether an error occurred or not, set with_error to True: it
will then return a tuple in the form of (json_response: any, error: bool).
url: Endpoint URL as string.
timeout: Connection timeout in seconds (0 for inf).
with_error: Return a tuple in the form (json_response: any, error: bool).
"""
return JSONConnector.request('GET', url, timeout=timeout,
with_error=with_error, **kwargs)
@staticmethod
def post(url: str, data: Union[str, dict], *, timeout: Union[int, float],
with_error: bool = False, **kwargs) -> Union[any, Tuple[any, bool]]:
"""Post data to a JSON endpoint, return the JSON converted response if any.
Request time, errors and exceptions are all logged using the standard
logger.
Note that the type of the returned response depends on the endpoint,
but it will always be some valid JSON or an empty dict.
To know whether an error occurred or not, set with_error to True: it
will then return a tuple in the form of (json_response: any, error: bool).
url: Endpoint URL as string.
data: Data to post, either as a dictionary (JSON valid) or a string.
timeout: Connection timeout in seconds (0 for inf).
with_error: Return a tuple in the form (json_response: any, error: bool).
"""
if isinstance(data, (dict, list)):
return JSONConnector.request('POST', url, timeout=timeout,
with_error=with_error, json=data, **kwargs)
return JSONConnector.request('POST', url, timeout=timeout,
with_error=with_error, data=data, **kwargs)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.