88 lines
3.9 KiB
Python
88 lines
3.9 KiB
Python
import logging
|
|
import typing
|
|
import json
|
|
import sys
|
|
from marshmallow import ValidationError
|
|
from werkzeug.exceptions import Forbidden
|
|
|
|
|
|
def create_default_json_response_format(error_field:str, error_description:str):
|
|
"""
|
|
The default format of a JSON response for exceptions is:
|
|
{
|
|
"error_field":str,
|
|
"error_description":str
|
|
}
|
|
"""
|
|
json_response = {
|
|
"error_field":error_field,
|
|
"error_description":error_description
|
|
}
|
|
return json_response
|
|
|
|
def unbundle_(errors, unbundled=[]):
|
|
"""
|
|
unbundles a dictionary entry into separate items and appends them to the list {unbundled}.
|
|
Example:
|
|
errors = {"key1":{"keya":"keyb","keyc":{"keyc12":12}}}
|
|
Returns:
|
|
[{'error_field':'keya', 'error_description':['keyb']}, {'error_field':'keyc12', 'error_description':[12]}]
|
|
As can be seen, only the subkeys and their respective value are received. Each value is *always* a list.
|
|
"""
|
|
{k:unbundle_(v,unbundled=unbundled) if isinstance(v,dict) else unbundled.append({"error_field":k, "error_description":v[0] if isinstance(v,list) else str(v)}) for k,v in errors.items()}
|
|
return
|
|
|
|
def unbundle_validation_error_message(message):
|
|
"""
|
|
inputs:
|
|
message: ValidationError.messages object. A str, list or dictionary
|
|
"""
|
|
unbundled = []
|
|
unbundle_(message, unbundled=unbundled)
|
|
if len(unbundled)>0:
|
|
error_field = "ValidationError in the following field(s): " + " & ".join([unb["error_field"] for unb in unbundled])
|
|
error_description = " " . join([unb["error_description"] for unb in unbundled])
|
|
else:
|
|
error_field = "ValidationError"
|
|
error_description = "unknown validation error"
|
|
return (error_field, error_description)
|
|
|
|
def create_validation_error_response(ex:ValidationError, status_code:int=400, create_log:bool=True)->typing.Tuple[str,int]:
|
|
# unbundles ValidationError into a dictionary of {'error_field':str, 'error_description':str}-format
|
|
message = ex.messages
|
|
(error_field, error_description) = unbundle_validation_error_message(message)
|
|
json_response = create_default_json_response_format(error_field=error_field, error_description=error_description)
|
|
|
|
# json.dumps with default=str automatically converts non-serializable values to strings. Hence, datetime objects (which are not)
|
|
# natively serializable are properly serialized.
|
|
serialized_response = json.dumps(json_response, default=str)
|
|
|
|
if create_log:
|
|
logging.warning(ex) if ex is not None else logging.warning(message)
|
|
# print(ex) if ex is not None else print(message)
|
|
return (serialized_response, status_code)
|
|
|
|
def create_werkzeug_error_response(ex:Forbidden, status_code:int=403, create_log:bool=True)->typing.Tuple[str,int]:
|
|
# json.dumps with default=str automatically converts non-serializable values to strings. Hence, datetime objects (which are not)
|
|
# natively serializable are properly serialized.
|
|
message = ex.description
|
|
json_response = create_default_json_response_format(error_field=str(repr(ex)), error_description=message)
|
|
serialized_response = json.dumps(json_response, default=str)
|
|
|
|
if create_log:
|
|
logging.warning(ex) if ex is not None else logging.warning(message)
|
|
# print(ex) if ex is not None else print(message)
|
|
return serialized_response, status_code
|
|
|
|
def create_dynamic_exception_response(ex, status_code:int=400, message:typing.Optional[str]=None, create_log:bool=True):
|
|
message = repr(ex) if message is None else message
|
|
json_response = create_default_json_response_format(error_field="Exception", error_description=message)
|
|
json_response["error_field"] = "call failed"
|
|
|
|
serialized_response = json.dumps(json_response, default=str)
|
|
|
|
if create_log:
|
|
logging.warning(ex) if ex is not None else logging.warning(message)
|
|
# print(ex) if ex is not None else print(message)
|
|
return (serialized_response, status_code)
|