phdi.transport

10def http_request_with_retry(
11    url: str,
12    retry_count: int,
13    request_type: Literal["GET", "POST"],
14    allowed_methods: List[str],
15    headers: dict,
16    data: dict = None,
17) -> requests.Response:
18    """
19    Executes an HTTP request, retrying the request if the returned HTTP status code
20    is one of a specified list of codes.
21
22    :param url: The url at which to make the HTTP request.
23    :param retry_count: The number of times to retry the request, if the
24      first attempt fails.
25    :param request_type: The type of request to be made. Currently supports
26      GET and POST.
27    :param allowed_methods: The list of allowed HTTP request methods (i.e.,
28      POST, PUT) for the specific URL and query.
29    :param headers: JSON-type dictionary of headers to make the request with,
30      including Authorization and content-type.
31    :param data: The data as a JSON-formatted dictionary, used when the request
32      requires data to be posted. Default: `None`
33    :raises ValueError: An unsupported HTTP method (e.g., PATCH, DELETE) was passed
34      to the request_type parameter.
35    :return: A HTTP request response.
36    """
37
38    request_type = request_type.upper()
39    if request_type not in ["GET", "POST"]:
40        raise ValueError(
41            f"The HTTP '{request_type}' method is not currently supported."
42        )
43
44    # Configure the settings of the 'requests' session we'll make
45    # the API call with
46    retry_strategy = Retry(
47        total=retry_count,
48        status_forcelist=[429, 500, 502, 503, 504],
49        allowed_methods=allowed_methods,
50    )
51    adapter = HTTPAdapter(max_retries=retry_strategy)
52    http = requests.Session()
53    http.mount("http://", adapter)
54    http.mount("https://", adapter)
55
56    # Now, actually try to complete the API request
57    # TODO: Condense this down to make a single call using
58    # http.request(method=request_type, url=url, headers=headers, json=data)
59    if request_type == "POST":
60        response = http.post(
61            url=url,
62            headers=headers,
63            json=data,
64        )
65    elif request_type == "GET":
66        response = http.get(
67            url=url,
68            headers=headers,
69        )
70
71    return response