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.