File: //home/arjun/projects/buyercall/buyercall/lib/util_requests.py
import requests
import json
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from flask import current_app
def requests_retry_session(retries=3, backoff_factor=0.3, status_forcelist=(500, 502, 504), session=None):
""" Returns a requests session with retry options
Args:
retries (int, optional): Defaults to 3.
backoff_factor (float, optional): A backoff factor to apply between attempts after the second try. Defaults to 0.3.
status_forcelist (tuple, optional): Defaults to (500, 502, 504).
session ([type], optional): Defaults to None.
Returns:
session_object: Request session
"""
session = session or requests.Session()
retry = Retry(
total=retries,
read=retries,
connect=retries,
backoff_factor=backoff_factor,
status_forcelist=status_forcelist,
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
def send_request(url='', payload=None, headers={}):
url = f"{current_app.config.get('SOCKET_ENDPOINT')}{url}"
try:
r = requests_retry_session().post(url, data=json.dumps(payload), headers=headers)
return (str(r.elapsed.total_seconds()), str(r.status_code), str(r.reason), r.content)
except:
pass
return True