Files @ f5ed6b020878
Branch filter:

Location: ChaosStuff/stockcli/stockcli/utils.py

Dennis Fink
Don't fail if the book has not isbn
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)