+# 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):
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):
elif domain == "":
raise ValueError("Parameter 'domain' is empty")
elif not isinstance(path, str):
elif domain == "":
raise ValueError("Parameter 'domain' is empty")
elif not isinstance(path, str):
elif path == "":
raise ValueError("Parameter 'path' cannot be empty")
elif not isinstance(data, str):
elif path == "":
raise ValueError("Parameter 'path' cannot be empty")
elif not isinstance(data, str):
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
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
# 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)}'"
# 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)}'"
if not isinstance(url, str):
raise ValueError(f"Parameter url[]='{type(url)}' is not 'str'")
elif not isinstance(timeout, tuple):
if not isinstance(url, str):
raise ValueError(f"Parameter url[]='{type(url)}' is not 'str'")
elif not isinstance(timeout, tuple):
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
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
# 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)}'"
# 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)}'"
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):
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):
elif domain == "":
raise ValueError("Parameter 'domain' is empty")
elif not isinstance(path, str):
elif domain == "":
raise ValueError("Parameter 'domain' is empty")
elif not isinstance(path, str):
elif path == "":
raise ValueError("Parameter 'path' cannot be empty")
elif not isinstance(headers, dict):
elif path == "":
raise ValueError("Parameter 'path' cannot be empty")
elif not isinstance(headers, dict):
# 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)}'"
# 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)}'"
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
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
instances.update_last_error(domain, response)
# DEBUG: print(f"DEBUG: Returning json_reply({len(json_reply)})=[]:{type(json_reply)}")
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):
elif path == "":
raise ValueError("Parameter 'path' is empty")
elif not isinstance(headers, dict):
# DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
instances.update_last_error(domain, exception)
raise exception
# DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
instances.update_last_error(domain, exception)
raise exception
if not isinstance(response, requests.models.Response):
raise ValueError(f"Parameter response[]='{type(response)}' is not type of 'Response'")
if not isinstance(response, requests.models.Response):
raise ValueError(f"Parameter response[]='{type(response)}' is not type of 'Response'")