Migrated from main API
This commit is contained in:
75
modules/utils.py
Normal file
75
modules/utils.py
Normal file
@@ -0,0 +1,75 @@
|
||||
from typing import Any, Union
|
||||
from ujson import loads, dumps, JSONDecodeError
|
||||
from traceback import print_exc
|
||||
|
||||
# Print to stdout and then to log
|
||||
def logWrite(message: str, debug: bool = False) -> None:
|
||||
# save to log file and rotation is to be done
|
||||
# logAppend(f'{message}', debug=debug)
|
||||
print(f"{message}", flush=True)
|
||||
|
||||
def jsonLoad(filepath: str) -> Any:
|
||||
"""Load json file
|
||||
|
||||
### Args:
|
||||
* filepath (`str`): Path to input file
|
||||
|
||||
### Returns:
|
||||
* `Any`: Some json deserializable
|
||||
"""
|
||||
with open(filepath, "r", encoding='utf8') as file:
|
||||
try:
|
||||
output = loads(file.read())
|
||||
except JSONDecodeError:
|
||||
logWrite(f"Could not load json file {filepath}: file seems to be incorrect!\n{print_exc()}")
|
||||
raise
|
||||
except FileNotFoundError:
|
||||
logWrite(f"Could not load json file {filepath}: file does not seem to exist!\n{print_exc()}")
|
||||
raise
|
||||
file.close()
|
||||
return output
|
||||
|
||||
def jsonSave(contents: Union[list, dict], filepath: str) -> None:
|
||||
"""Save contents into json file
|
||||
|
||||
### Args:
|
||||
* contents (`Union[list, dict]`): Some json serializable
|
||||
* filepath (`str`): Path to output file
|
||||
"""
|
||||
try:
|
||||
with open(filepath, "w", encoding='utf8') as file:
|
||||
file.write(dumps(contents, ensure_ascii=False, indent=4))
|
||||
file.close()
|
||||
except Exception as exp:
|
||||
logWrite(f"Could not save json file {filepath}: {exp}\n{print_exc()}")
|
||||
return
|
||||
|
||||
def configGet(key: str, *args: str) -> Any:
|
||||
"""Get value of the config key
|
||||
|
||||
### Args:
|
||||
* key (`str`): The last key of the keys path.
|
||||
* *args (`str`): Path to key like: dict[args][key].
|
||||
|
||||
### Returns:
|
||||
* `Any`: Value of provided key
|
||||
"""
|
||||
this_dict = jsonLoad("config.json")
|
||||
this_key = this_dict
|
||||
for dict_key in args:
|
||||
this_key = this_key[dict_key]
|
||||
return this_key[key]
|
||||
|
||||
def apiKeyInvalid(obj):
|
||||
obj.send_response(401)
|
||||
obj.send_header('Content-type', 'application/json; charset=utf-8')
|
||||
obj.end_headers()
|
||||
obj.wfile.write(b'{"code":401, "message": "Invalid API key"}')
|
||||
return
|
||||
|
||||
def apiKeyExpired(obj):
|
||||
obj.send_response(403)
|
||||
obj.send_header('Content-type', 'application/json; charset=utf-8')
|
||||
obj.end_headers()
|
||||
obj.wfile.write(b'{"code":403, "message": "API key expired"}')
|
||||
return
|
Reference in New Issue
Block a user