error response due to language models are updated
This commit is contained in:
@@ -1,39 +1,57 @@
|
||||
from typing import Any, Dict
|
||||
from typing import Any, Dict, Union, Awaitable
|
||||
from fastapi import Request, WebSocket
|
||||
from fastapi.responses import Response
|
||||
|
||||
from LanguageModels.Errors.merge_all_error_languages import MergedErrorLanguageModels
|
||||
from ErrorHandlers.Exceptions.api_exc import HTTPExceptionApi
|
||||
from ErrorHandlers.bases import BaseErrorModelClass
|
||||
|
||||
|
||||
class HTTPExceptionApiHandler:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
**kwargs,
|
||||
response_model: Any,
|
||||
):
|
||||
self.EXCEPTIONS = kwargs.get(
|
||||
"exceptions"
|
||||
) # from fastapi.exceptions import HTTPException
|
||||
self.STATUSES = kwargs.get("statuses") # from fastapi import status
|
||||
self.EXCEPTION_DICTS: Dict = kwargs.get("exceptions_dict")
|
||||
self.ERRORS_DICT: Dict = kwargs.get("errors_dict")
|
||||
self.ERRORS_LANG: Dict = kwargs.get("error_language_dict")
|
||||
self.RESPONSE_MODEL: Any = kwargs.get("response_model")
|
||||
self.RESPONSE_MODEL: Any = response_model
|
||||
|
||||
def retrieve_error_status_code(self, exc: HTTPExceptionApi):
|
||||
grab_status = self.ERRORS_DICT.get(str(exc.error_code).upper(), "")
|
||||
grab_status_code = self.EXCEPTION_DICTS.get(str(grab_status).upper(), "500")
|
||||
return getattr(
|
||||
self.STATUSES,
|
||||
str(grab_status_code),
|
||||
getattr(self.STATUSES, "HTTP_500_INTERNAL_SERVER_ERROR"),
|
||||
@staticmethod
|
||||
def retrieve_error_status_code(exc: HTTPExceptionApi) -> int:
|
||||
from ErrorHandlers import DEFAULT_ERROR
|
||||
error_by_codes = BaseErrorModelClass.retrieve_error_by_codes()
|
||||
grab_status_code = error_by_codes.get(
|
||||
str(exc.error_code).upper(), DEFAULT_ERROR
|
||||
)
|
||||
return int(grab_status_code)
|
||||
|
||||
def retrieve_error_message(self, exc: HTTPExceptionApi):
|
||||
message_by_lang = self.ERRORS_LANG.get(str(exc.lang).lower(), {})
|
||||
return message_by_lang.get(str(exc.error_code).upper(), "Unknown error")
|
||||
@staticmethod
|
||||
def retrieve_error_message(exc: HTTPExceptionApi, error_languages) -> str:
|
||||
from ErrorHandlers import DEFAULT_ERROR
|
||||
return error_languages.get(str(exc.error_code).upper(), DEFAULT_ERROR)
|
||||
|
||||
def handle_exception(self, request, exc: HTTPExceptionApi):
|
||||
status_code = self.retrieve_error_status_code(exc)
|
||||
error_message = self.retrieve_error_message(exc)
|
||||
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,
|
||||
},
|
||||
)
|
||||
return self.RESPONSE_MODEL(
|
||||
status_code=int(status_code),
|
||||
content={"message": error_message, "lang": exc.lang, "request": request},
|
||||
)
|
||||
status_code=500,
|
||||
content={
|
||||
"message": "Internal Server Error",
|
||||
"lang": "def",
|
||||
"request": request_string,
|
||||
},
|
||||
) # Handle other exceptions with a generic 500 error
|
||||
|
||||
@@ -4,9 +4,10 @@ from ErrorHandlers.ErrorHandlers.api_exc_handler import (
|
||||
from ErrorHandlers.Exceptions.api_exc import (
|
||||
HTTPExceptionApi,
|
||||
)
|
||||
|
||||
DEFAULT_ERROR = "UNKNOWN_ERROR"
|
||||
|
||||
__all__ = [
|
||||
"HTTPExceptionApiHandler",
|
||||
"HTTPExceptionApi",
|
||||
"DEFAULT_ERROR"
|
||||
]
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
from ErrorHandlers.bases import BaseErrorModelClass
|
||||
|
||||
|
||||
class BaseError(BaseErrorModelClass):
|
||||
class BaseError:
|
||||
NOT_CREATED: int = 405
|
||||
NOT_DELETED: int = 405
|
||||
NOT_UPDATED: int = 405
|
||||
@@ -14,3 +11,4 @@ class BaseError(BaseErrorModelClass):
|
||||
NOT_ACCEPTABLE: int = 406
|
||||
INVALID_DATA: int = 422
|
||||
UNKNOWN_ERROR: int = 502
|
||||
|
||||
|
||||
@@ -1,30 +1,19 @@
|
||||
from typing import Optional
|
||||
from ErrorHandlers.base import BaseError
|
||||
from ErrorHandlers.statuses import Statuses
|
||||
|
||||
|
||||
class BaseErrorModelClass:
|
||||
list_of_statuses = [Statuses, BaseError]
|
||||
|
||||
@classmethod
|
||||
def retrieve_error_by_code(cls, error_code: str):
|
||||
return getattr(cls, error_code, 502)
|
||||
def retrieve_error_by_codes(cls):
|
||||
language_model_status = {}
|
||||
for list_of_language in cls.list_of_statuses:
|
||||
clean_dict = {
|
||||
key: value
|
||||
for key, value in list_of_language.__dict__.items()
|
||||
if "__" not in str(key)[0:3]
|
||||
}
|
||||
language_model_status.update(clean_dict)
|
||||
return language_model_status
|
||||
|
||||
|
||||
class StatusesModelClass:
|
||||
|
||||
@classmethod
|
||||
def retrieve_error_by_code(cls, error_code: str):
|
||||
return getattr(cls, error_code, 502)
|
||||
|
||||
|
||||
class ErrorLanguageModelClass:
|
||||
|
||||
@classmethod
|
||||
def retrieve_error_header(cls, error_code: str):
|
||||
return getattr(cls, error_code, "Unknown Error occured.")
|
||||
|
||||
|
||||
class LanguageModelClass:
|
||||
|
||||
@classmethod
|
||||
def retrieve_error_by_code(cls, error_code: str, language: Optional[str] = "tr"):
|
||||
language_model: ErrorLanguageModelClass = getattr(cls, language, "tr")
|
||||
return language_model.retrieve_error_header(error_code)
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
from ErrorHandlers.bases import StatusesModelClass
|
||||
|
||||
|
||||
class Statuses(StatusesModelClass):
|
||||
class Statuses:
|
||||
HTTP_100_CONTINUE = 100
|
||||
HTTP_101_SWITCHING_PROTOCOLS = 101
|
||||
HTTP_102_PROCESSING = 102
|
||||
|
||||
Reference in New Issue
Block a user