afscgap.http_util

Utilities for HTTP requests as part of afscgap.

(c) 2023 Regents of University of California / The Eric and Wendy Schmidt Center for Data Science and the Environment at UC Berkeley.

This file is part of afscgap released under the BSD 3-Clause License. See LICENSE.md.

 1"""
 2Utilities for HTTP requests as part of afscgap.
 3
 4(c) 2023 Regents of University of California / The Eric and Wendy Schmidt Center
 5for Data Science and the Environment at UC Berkeley.
 6
 7This file is part of afscgap released under the BSD 3-Clause License. See
 8LICENSE.md.
 9"""
10import requests
11
12from afscgap.typesdef import REQUESTOR
13
14TIMEOUT = 60 * 5  # 5 minutes
15
16
17def check_result(target: requests.Response):
18    """Assert that a result returned an acceptable status code.
19
20    Args:
21        target: The response to check.
22
23    Raises:
24        RuntimeError: Raised if the response returned indicates an issue or
25            unexpected status code.
26    """
27    status_ok = target.status_code >= 100 and target.status_code < 400
28    if not status_ok:
29        message = 'Got non-OK response from remote: %d (%s)' % (
30            target.status_code,
31            target.text
32        )
33        raise RuntimeError(message)
34
35
36def build_requestor(stream: bool = False) -> REQUESTOR:
37    """Build a requestor strategy that uses the requests library.
38
39    Returns:
40        Newly built strategy.
41    """
42    return lambda x: requests.get(x, timeout=TIMEOUT, stream=stream)
TIMEOUT = 300
def check_result(target: requests.models.Response):
18def check_result(target: requests.Response):
19    """Assert that a result returned an acceptable status code.
20
21    Args:
22        target: The response to check.
23
24    Raises:
25        RuntimeError: Raised if the response returned indicates an issue or
26            unexpected status code.
27    """
28    status_ok = target.status_code >= 100 and target.status_code < 400
29    if not status_ok:
30        message = 'Got non-OK response from remote: %d (%s)' % (
31            target.status_code,
32            target.text
33        )
34        raise RuntimeError(message)

Assert that a result returned an acceptable status code.

Arguments:
  • target: The response to check.
Raises:
  • RuntimeError: Raised if the response returned indicates an issue or unexpected status code.
def build_requestor(stream: bool = False) -> Callable[[str], requests.models.Response]:
37def build_requestor(stream: bool = False) -> REQUESTOR:
38    """Build a requestor strategy that uses the requests library.
39
40    Returns:
41        Newly built strategy.
42    """
43    return lambda x: requests.get(x, timeout=TIMEOUT, stream=stream)

Build a requestor strategy that uses the requests library.

Returns:

Newly built strategy.