90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
import json
|
|
from typing import Any, Union, Awaitable
|
|
from pydantic import ValidationError
|
|
|
|
from fastapi import Request, WebSocket, status
|
|
from fastapi.responses import Response, JSONResponse
|
|
from ApiLayers.LanguageModels.Errors.merge_all_error_languages import (
|
|
MergedErrorLanguageModels,
|
|
)
|
|
from ApiLayers.ErrorHandlers.Exceptions.api_exc import HTTPExceptionApi
|
|
from ApiLayers.ErrorHandlers.bases import BaseErrorModelClass
|
|
|
|
|
|
def validation_exception_handler(request, exc: ValidationError) -> JSONResponse:
|
|
"""
|
|
{"message": [{
|
|
"type": "missing", "location": ["company_uu_id"], "message": "Field required",
|
|
"input": {"invalid_key_input": "e9869a25"}
|
|
}], "request": "http://0.0.0.0:41575/authentication/select", "title": "EmployeeSelection"
|
|
}
|
|
Validation error on pydantic model of each event validation
|
|
"""
|
|
validation_messages, validation_list = exc.errors() or [], []
|
|
for validation in validation_messages:
|
|
validation_list.append(
|
|
{
|
|
"type": dict(validation).get("type"),
|
|
"location": dict(validation).get("loc"),
|
|
"message": dict(validation).get(
|
|
"msg"
|
|
), # todo change message with language message
|
|
"input": dict(validation).get("input"),
|
|
}
|
|
)
|
|
error_response_dict = dict(
|
|
message=validation_list, request=str(request.url.path), title=exc.title
|
|
)
|
|
return JSONResponse(
|
|
content=error_response_dict, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY
|
|
)
|
|
|
|
|
|
class HTTPExceptionApiHandler:
|
|
|
|
def __init__(self, response_model: Any):
|
|
self.RESPONSE_MODEL: Any = response_model
|
|
|
|
@staticmethod
|
|
def retrieve_error_status_code(exc: HTTPExceptionApi) -> int:
|
|
error_by_codes = BaseErrorModelClass.retrieve_error_by_codes()
|
|
grab_status_code = error_by_codes.get(str(exc.error_code).upper(), 500)
|
|
return int(grab_status_code)
|
|
|
|
@staticmethod
|
|
def retrieve_error_message(exc: HTTPExceptionApi, error_languages) -> str:
|
|
from ApiLayers.ErrorHandlers import DEFAULT_ERROR
|
|
|
|
return error_languages.get(str(exc.error_code).upper(), DEFAULT_ERROR)
|
|
|
|
async def handle_exception(
|
|
self, request: Union[Request, WebSocket], exc: Exception
|
|
) -> Union[Response, Awaitable[None]]:
|
|
request_string = (
|
|
str(request.url) if isinstance(request, Request) else request.url.path
|
|
)
|
|
if isinstance(exc, HTTPExceptionApi):
|
|
error_languages = MergedErrorLanguageModels.get_language_models(
|
|
language=exc.lang
|
|
)
|
|
status_code = self.retrieve_error_status_code(exc)
|
|
error_message = self.retrieve_error_message(exc, error_languages)
|
|
return self.RESPONSE_MODEL(
|
|
status_code=int(status_code),
|
|
content={
|
|
"message": error_message,
|
|
"lang": exc.lang,
|
|
"request": request_string,
|
|
"loc": exc.loc,
|
|
},
|
|
)
|
|
return self.RESPONSE_MODEL(
|
|
status_code=500,
|
|
content={
|
|
"message": "Internal Server Error",
|
|
"lang": "def",
|
|
"request": request_string,
|
|
"loc": exc.loc,
|
|
},
|
|
) # Handle other exceptions with a generic 500 error
|