"Content-Type": "application/json",
}
+# Exceptions to always catch
+exceptions = (
+ requests.exceptions.ConnectionError,
+ requests.exceptions.Timeout,
+ requests.exceptions.TooManyRedirects,
+ UnicodeEncodeError
+)
+
def post_json_api(domain: str, path: str, data: str, headers: dict = {}) -> dict:
# DEBUG: print(f"DEBUG: domain='{domain}',path='{path}',data='{data}',headers()={len(headers)} - CALLED!")
if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
+ raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
elif domain == "":
raise ValueError("Parameter 'domain' is empty")
elif not isinstance(path, str):
- raise ValueError(f"path[]={type(path)} is not 'str'")
+ raise ValueError(f"path[]='{type(path)}' is not 'str'")
elif path == "":
raise ValueError("Parameter 'path' cannot be empty")
elif not isinstance(data, str):
- raise ValueError(f"data[]={type(data)} is not 'str'")
+ raise ValueError(f"data[]='{type(data)}' is not 'str'")
elif not isinstance(headers, dict):
- raise ValueError(f"headers[]={type(headers)} is not 'list'")
+ raise ValueError(f"headers[]='{type(headers)}' is not 'list'")
json_reply = {
"status_code": 200,
if not response.ok or response.status_code >= 400:
print(f"WARNING: Cannot query JSON API: domain='{domain}',path='{path}',data()={len(data)},response.status_code='{response.status_code}',json_reply[]='{type(json_reply)}'")
json_reply["status_code"] = response.status_code
- json_reply["error_message"] = response.text
+ json_reply["error_message"] = response.reason
+ del json_reply["json"]
instances.update_last_error(domain, response)
- except requests.exceptions.ConnectionError as exception:
+ except exceptions as exception:
# DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
json_reply["status_code"] = 999
json_reply["error_message"] = f"exception['{type(exception)}']='{str(exception)}'"
+ json_reply["exception"] = exception
instances.update_last_error(domain, exception)
raise exception
if not isinstance(url, str):
raise ValueError(f"Parameter url[]='{type(url)}' is not 'str'")
elif not isinstance(timeout, tuple):
- raise ValueError(f"timeout[]={type(timeout)} is not 'tuple'")
+ raise ValueError(f"timeout[]='{type(timeout)}' is not 'tuple'")
json_reply = {
"status_code": 200,
if not response.ok or response.status_code >= 400:
print(f"WARNING: Cannot query JSON API: url='{url}',response.status_code='{response.status_code}',json_reply[]='{type(json_reply)}'")
json_reply["status_code"] = response.status_code
- json_reply["error_message"] = response.text
+ json_reply["error_message"] = response.reason
+ del json_reply["json"]
- except requests.exceptions.ConnectionError as exception:
+ except exceptions as exception:
# DEBUG: print(f"DEBUG: Fetching '{url}' failed. exception[{type(exception)}]='{str(exception)}'")
json_reply["status_code"] = 999
json_reply["error_message"] = f"exception['{type(exception)}']='{str(exception)}'"
+ json_reply["exception"] = exception
raise exception
# DEBUG: print(f"DEBUG: Returning json_reply({len(json_reply)})=[]:{type(json_reply)}")
def get_json_api(domain: str, path: str, headers: dict, timeout: tuple) -> dict:
# DEBUG: print(f"DEBUG: domain='{domain}',path='{path}',timeout()={len(timeout)} - CALLED!")
if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
+ raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
elif domain == "":
raise ValueError("Parameter 'domain' is empty")
elif not isinstance(path, str):
- raise ValueError(f"path[]={type(path)} is not 'str'")
+ raise ValueError(f"path[]='{type(path)}' is not 'str'")
elif path == "":
raise ValueError("Parameter 'path' cannot be empty")
elif not isinstance(headers, dict):
- raise ValueError(f"headers[]={type(headers)} is not 'list'")
+ raise ValueError(f"headers[]='{type(headers)}' is not 'list'")
elif not isinstance(timeout, tuple):
- raise ValueError(f"timeout[]={type(timeout)} is not 'tuple'")
+ raise ValueError(f"timeout[]='{type(timeout)}' is not 'tuple'")
json_reply = {
"status_code": 200,
timeout=timeout
)
- except requests.exceptions.ConnectionError as exception:
+ except exceptions as exception:
# DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
json_reply["status_code"] = 999
json_reply["error_message"] = f"exception['{type(exception)}']='{str(exception)}'"
+ json_reply["exception"] = exception
instances.update_last_error(domain, exception)
raise exception
if not response.ok or response.status_code >= 400:
print(f"WARNING: Cannot query JSON API: domain='{domain}',path='{path}',response.status_code='{response.status_code}',json_reply[]='{type(json_reply)}'")
json_reply["status_code"] = response.status_code
- json_reply["error_message"] = response.text
+ json_reply["error_message"] = response.reason
+ del json_reply["json"]
instances.update_last_error(domain, response)
# DEBUG: print(f"DEBUG: Returning json_reply({len(json_reply)})=[]:{type(json_reply)}")
elif path == "":
raise ValueError("Parameter 'path' is empty")
elif not isinstance(headers, dict):
- raise ValueError(f"headers[]={type(headers)} is not 'dict'")
+ raise ValueError(f"headers[]='{type(headers)}' is not 'dict'")
elif not isinstance(timeout, tuple):
- raise ValueError(f"timeout[]={type(timeout)} is not 'tuple'")
+ raise ValueError(f"timeout[]='{type(timeout)}' is not 'tuple'")
try:
# DEBUG: print(f"DEBUG: Sending GET request to '{domain}{path}' ...")
timeout=timeout
)
- except requests.exceptions.ConnectionError as exception:
+ except exceptions as exception:
# DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
instances.update_last_error(domain, exception)
raise exception
return response
def json_from_response(response: requests.models.Response) -> list:
- # DEBUG: print(f"DEBUG: response[]={type(response)} - CALLED!")
+ # DEBUG: print(f"DEBUG: response[]='{type(response)}' - CALLED!")
if not isinstance(response, requests.models.Response):
raise ValueError(f"Parameter response[]='{type(response)}' is not type of 'Response'")
except json.decoder.JSONDecodeError:
pass
- # DEBUG: print(f"DEBUG: data[]={type(data)} - EXIT!")
+ # DEBUG: print(f"DEBUG: data[]='{type(data)}' - EXIT!")
return data