def set(key: str, domain: str, value: any):
# DEBUG: print(f"DEBUG: key='{key}',domain='{domain}',value[]='{type(value)}' - CALLED!")
- if type(key) != str:
+ if not isinstance(key, str):
raise ValueError("Parameter key[]='{type(key)}' is not 'str'")
elif key == "":
- raise ValueError(f"Parameter 'key' cannot be empty")
- elif type(domain) != str:
- raise ValueError("Parameter domain[]='{type(domain)}' is not 'str'")
+ raise ValueError("Parameter 'key' is empty")
+ elif not isinstance(domain, str):
+ raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' cannot be empty")
+ raise ValueError("Parameter 'domain' is empty")
elif not key in _pending:
raise ValueError(f"key='{key}' not found in _pending")
elif not fba.is_primitive(value):
def has_pending_instance_data(domain: str) -> bool:
# DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
+ raise ValueError("Parameter 'domain' is empty")
has_pending = False
for key in _pending:
def update_data(domain: str):
# DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
+ raise ValueError("Parameter 'domain' is empty")
elif not has_pending_instance_data(domain):
raise Exception(f"Domain '{domain}' has no pending instance data, but function invoked")
except:
pass
- except BaseException as e:
- print(f"ERROR: failed SQL query: domain='{domain}',sql_string='{sql_string}',exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: failed SQL query: domain='{domain}',sql_string='{sql_string}',exception[{type(exception)}]:'{str(exception)}'")
sys.exit(255)
# DEBUG: print("DEBUG: EXIT!")
def update_last_instance_fetch(domain: str):
# DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
+ raise ValueError("Parameter 'domain' is empty")
# DEBUG: print("DEBUG: Updating last_instance_fetch for domain:", domain)
set("last_instance_fetch", domain, time.time())
# DEBUG: print("DEBUG: EXIT!")
def update_last_blocked(domain: str):
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
+ raise ValueError("Parameter 'domain' is empty")
# DEBUG: print("DEBUG: Updating last_blocked for domain", domain)
set("last_blocked", domain, time.time())
def add(domain: str, origin: str, originator: str, path: str = None):
# DEBUG: print(f"DEBUG: domain='{domain}',origin='{origin}',originator='{originator}',path='{path}' - CALLED!")
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
- elif type(origin) != str and origin != None:
+ raise ValueError("Parameter 'domain' is empty")
+ elif not isinstance(origin, str) and origin is not None:
raise ValueError(f"origin[]={type(origin)} is not 'str'")
- elif type(originator) != str:
+ elif origin == "":
+ raise ValueError("Parameter 'origin' is empty")
+ elif not isinstance(originator, str):
raise ValueError(f"originator[]={type(originator)} is not 'str'")
elif originator == "":
- raise ValueError(f"originator cannot be empty")
+ raise ValueError("Parameter 'originator' is empty")
elif not validators.domain(domain.split("/")[0]):
raise ValueError(f"Bad domain name='{domain}'")
elif origin is not None and not validators.domain(origin.split("/")[0]):
update_last_error(domain, fba.pending_errors[domain])
fba.remove_pending_error(domain)
- except BaseException as e:
- print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(exception)}]:'{str(exception)}'")
sys.exit(255)
else:
# DEBUG: print("DEBUG: Updating nodeinfo for domain:", domain)
def update_last_nodeinfo(domain: str):
# DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
+ raise ValueError("Parameter 'domain' is empty")
# DEBUG: print("DEBUG: Updating last_nodeinfo for domain:", domain)
set("last_nodeinfo", domain, time.time())
def update_last_error(domain: str, response: requests.models.Response):
# DEBUG: print("DEBUG: domain,response[]:", domain, type(response))
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
+ raise ValueError("Parameter 'domain' is empty")
# DEBUG: print("DEBUG: BEFORE response[]:", type(response))
if isinstance(response, BaseException) or isinstance(response, json.decoder.JSONDecodeError):
response = f"{type}:str(response)"
# DEBUG: print("DEBUG: AFTER response[]:", type(response))
- if type(response) is str:
+ if isinstance(response, str):
# DEBUG: print(f"DEBUG: Setting last_error_details='{response}'");
set("last_status_code" , domain, 999)
set("last_error_details", domain, response)
def is_registered(domain: str) -> bool:
# DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
+ raise ValueError("Parameter 'domain' is empty")
# DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
if not cache.key_exists("is_registered"):
# Check Set all
cache.set_all("is_registered", fba.cursor.fetchall(), True)
- except BaseException as e:
- print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(exception)}]:'{str(exception)}'")
sys.exit(255)
# Is cache found?