import logging
import string
from operator import attrgetter
from typing import Any, Dict, Optional
import click
import requests
from .console import error_console
UNALLOWED_CHARACTERS = str.maketrans(dict((c, None) for c in string.whitespace))
def make_request(
method: str, url_path: str, data: Optional[Any] = None, *, cached: bool = False
) -> Any:
obj = click.get_current_context().obj
if cached:
session = obj["cached_session"]
else:
session = obj["request_session"]
base_url = obj["base_url"]
requested_url = base_url + url_path
method_function = attrgetter(method)
try:
if data is not None:
logging.debug(
f"Making {method.upper()} request: {requested_url} with {data}"
)
response = method_function(session)(requested_url, json=data)
else:
logging.debug(f"Making {method.upper()} request: {requested_url}")
response = method_function(session)(requested_url)
response.raise_for_status()
except requests.Timeout:
logging.error(f"The connection to {requested_url} timed out!")
error_console.print("Connection timed out!")
raise
except requests.ConnectionError:
logging.error(f"Couldn't establish a connection to {requested_url}!")
error_console.print("Couldn't establish a connection!")
raise
except requests.HTTPError:
logging.error(f"{requested_url} sent back an HTTPError")
error_console.print("Got the following error:")
error_message = response.json()["error_message"]
logging.error(error_message)
error_console.print(error_message)
raise
except requests.TooManyRedirects:
logging.error(f"{requested_url} had too many redirects!")
error_console.print("Too many redirects!")
raise
else:
try:
return response.json()
except requests.JSONDecodeError:
return response
def get_request(url_path: str, *, cached: bool = False) -> Any:
return make_request("get", url_path, cached=cached)
def post_request(url_path: str, data: Dict[str, Any], *, cached: bool = False) -> Any:
return make_request("post", url_path, data, cached=cached)
def put_request(url_path: str, data: Dict[str, Any], *, cached: bool = False) -> Any:
return make_request("put", url_path, data, cached=cached)
def prepare_barcode(barcode: str) -> str:
return barcode.translate(UNALLOWED_CHARACTERS)