Moved from plain text error response to JSON response

Author: eadwinCode

ellar/core/exceptions/__init__.py

@@ -3,10 +3,13 @@
 from starlette import status
 from starlette.exceptions import HTTPException, WebSocketException
 
-from .base import APIException, ErrorDetail
+from .base import APIException
+from .interfaces import IExceptionHandler, IExceptionMiddlewareService
 from .validation import RequestValidationError, WebSocketRequestValidationError
 
 __all__ = [
+    "IExceptionHandler",
+    "IExceptionMiddlewareService",
     "HTTPException",
     "ImproperConfiguration",
     "APIException",
@@ -29,70 +32,41 @@ class ImproperConfiguration(Exception):
 
 class AuthenticationFailed(APIException):
     status_code = status.HTTP_401_UNAUTHORIZED
-    default_detail = "Incorrect authentication credentials."
-    default_code = "authentication_failed"
+    code = "authentication_failed"
+
+    def __init__(
+        self, detail: t.Union[t.List, t.Dict, str] = None, **kwargs: t.Any
+    ) -> None:
+        if detail is None:
+            detail = "Incorrect authentication credentials."
+        super(AuthenticationFailed, self).__init__(detail=detail, **kwargs)
 
 
 class NotAuthenticated(APIException):
     status_code = status.HTTP_401_UNAUTHORIZED
-    default_detail = "Authentication credentials were not provided."
-    default_code = "not_authenticated"
+    code = "not_authenticated"
 
 
 class PermissionDenied(APIException):
     status_code = status.HTTP_403_FORBIDDEN
-    default_detail = "You do not have permission to perform this action."
-    default_code = "permission_denied"
+    code = "permission_denied"
 
 
 class NotFound(APIException):
     status_code = status.HTTP_404_NOT_FOUND
-    default_detail = "Not found."
-    default_code = "not_found"
+    code = "not_found"
 
 
 class MethodNotAllowed(APIException):
     status_code = status.HTTP_405_METHOD_NOT_ALLOWED
-    default_detail = 'Method "{method}" not allowed.'
-    default_code = "method_not_allowed"
-
-    def __init__(
-        self,
-        method: str,
-        detail: t.Optional[t.Union[t.List, t.Dict, ErrorDetail, str]] = None,
-        code: t.Optional[t.Union[str, int]] = None,
-    ):
-        if detail is None:
-            detail = str(self.default_detail).format(method=method)
-        super().__init__(detail, code)
+    code = "method_not_allowed"
 
 
 class NotAcceptable(APIException):
     status_code = status.HTTP_406_NOT_ACCEPTABLE
-    default_detail = "Could not satisfy the request Accept header."
-    default_code = "not_acceptable"
-
-    def __init__(
-        self,
-        detail: t.Optional[t.Union[t.List, t.Dict, ErrorDetail, str]] = None,
-        code: t.Optional[t.Union[str, int]] = None,
-        available_renderers: t.Optional[str] = None,
-    ):
-        self.available_renderers = available_renderers
-        super().__init__(detail, code)
+    code = "not_acceptable"
 
 
 class UnsupportedMediaType(APIException):
     status_code = status.HTTP_415_UNSUPPORTED_MEDIA_TYPE
-    default_detail = 'Unsupported media type "{media_type}" in request.'
-    default_code = "unsupported_media_type"
-
-    def __init__(
-        self,
-        media_type: str,
-        detail: t.Optional[t.Union[t.List, t.Dict, ErrorDetail, str]] = None,
-        code: t.Optional[t.Union[str, int]] = None,
-    ):
-        if detail is None:
-            detail = str(self.default_detail).format(media_type=media_type)
-        super().__init__(detail, code)
+    code = "unsupported_media_type"

ellar/core/exceptions/base.py

@@ -1,128 +1,49 @@
+import http
 import typing as t
 
 from starlette import status
-from starlette.exceptions import HTTPException as StarletteHTTPException
 
 
-@t.no_type_check
-def _get_error_details(
-    data: t.Union[t.List, t.Dict, "ErrorDetail"],
-    default_code: t.Optional[t.Union[str, int]] = None,
-) -> t.Union[t.List["ErrorDetail"], "ErrorDetail", t.Dict[t.Any, "ErrorDetail"]]:
-    """
-    Descend into a nested data structure, forcing any
-    lazy translation strings or strings into `ErrorDetail`.
-    """
-    if isinstance(data, list):
-        ret = [_get_error_details(item, default_code) for item in data]
-        return ret
-    elif isinstance(data, dict):
-        ret = {
-            key: _get_error_details(value, default_code) for key, value in data.items()
-        }
-        return ret
+class APIException(Exception):
+    __slots__ = ("headers", "description", "detail", "http_status")
 
-    text = str(data)
-    code = getattr(data, "code", default_code)
-    return ErrorDetail(text, code)
-
-
-@t.no_type_check
-def _get_codes(
-    detail: t.Union[t.List, t.Dict, "ErrorDetail"]
-) -> t.Union[str, t.Dict, t.List[t.Dict]]:
-    if isinstance(detail, list):
-        return [_get_codes(item) for item in detail]
-    elif isinstance(detail, dict):
-        return {key: _get_codes(value) for key, value in detail.items()}
-    return detail.code
-
-
-@t.no_type_check
-def _get_full_details(
-    detail: t.Union[t.List, t.Dict, "ErrorDetail"]
-) -> t.Union[t.Dict, t.List[t.Dict]]:
-    if isinstance(detail, list):
-        return [_get_full_details(item) for item in detail]
-    elif isinstance(detail, dict):
-        return {key: _get_full_details(value) for key, value in detail.items()}
-    return {"message": detail, "code": detail.code}
-
-
-class ErrorDetail(str):
-    """
-    A string-like object that can additionally have a code.
-    """
-
-    code = None
-
-    def __new__(
-        cls, string: str, code: t.Optional[t.Union[str, int]] = None
-    ) -> "ErrorDetail":
-        self = super().__new__(cls, string)
-        self.code = code
-        return self
-
-    def __eq__(self, other: object) -> bool:
-        r = super().__eq__(other)
-        try:
-            return r and self.code == other.code  # type: ignore
-        except AttributeError:
-            return r
-
-    def __ne__(self, other: object) -> bool:
-        return not self.__eq__(other)
-
-    def __repr__(self) -> str:
-        return "ErrorDetail(string=%r, code=%r)" % (
-            str(self),
-            self.code,
-        )
-
-    def __hash__(self) -> t.Any:
-        return hash(str(self))
-
-
-class APIException(StarletteHTTPException):
-    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-    default_detail = "A server error occurred."
-    default_code = "error"
+    status_code = status.HTTP_400_BAD_REQUEST
+    code = "bad_request"
 
     def __init__(
         self,
-        detail: t.Optional[t.Union[t.List, t.Dict, "ErrorDetail", str]] = None,
-        code: t.Optional[t.Union[str, int]] = None,
-        status_code: t.Optional[int] = None,
-        headers: t.Optional[t.Dict[str, t.Any]] = None,
+        detail: t.Union[t.List, t.Dict, str] = None,
+        description: str = None,
+        headers: t.Dict[str, t.Any] = None,
+        status_code: int = None,
     ) -> None:
-        if detail is None:
-            detail = self.default_detail
-        if code is None:
-            code = self.default_code
+        assert self.status_code
+        self.status_code = status_code or self.status_code
+        self.http_status = http.HTTPStatus(self.status_code)
+        self.description = description
 
-        if status_code is None:
-            status_code = self.status_code
+        if detail is None:
+            detail = self.http_status.phrase
 
-        super(APIException, self).__init__(
-            status_code=status_code,
-            detail=_get_error_details(detail, code),
-            headers=headers,
-        )
+        self.detail = detail
+        self.headers = headers
 
-    def get_codes(
-        self,
-    ) -> t.Union[str, t.Dict, t.List]:
-        """
-        Return only the code part of the error details.
-        Eg. {"name": ["required"]}
-        """
-        return _get_codes(t.cast(ErrorDetail, self.detail))  # type: ignore
+    def __repr__(self) -> str:
+        class_name = self.__class__.__name__
+        return f"{class_name}(status_code={self.status_code!r}, detail={self.detail!r})"
 
-    def get_full_details(
-        self,
-    ) -> t.Union[t.Dict, t.List[t.Dict]]:
+    def get_full_details(self) -> t.Union[t.Dict, t.List[t.Dict]]:
         """
         Return both the message & code parts of the error details.
-        Eg. {"name": [{"message": "This field is required.", "code": "required"}]}
         """
-        return _get_full_details(t.cast(ErrorDetail, self.detail))  # type: ignore
+        return dict(
+            detail=self.detail,
+            code=self.code,
+            description=self.description or self.http_status.description,
+        )
+
+    def get_details(self) -> t.Dict:
+        result = dict(detail=self.detail)
+        if self.description:
+            result.update(description=self.description)
+        return result

ellar/core/exceptions/handlers.py

@@ -5,7 +5,7 @@
     HTTPException as StarletteHTTPException,
     WebSocketException as StarletteWebSocketException,
 )
-from starlette.responses import PlainTextResponse, Response
+from starlette.responses import Response
 
 from ellar.core.context import IExecutionContext
 from ellar.core.exceptions import APIException, RequestValidationError
@@ -20,10 +20,19 @@ class HTTPExceptionHandler(IExceptionHandler):
     async def catch(
         self, ctx: IExecutionContext, exc: StarletteHTTPException
     ) -> t.Union[Response, t.Any]:
+        assert isinstance(exc, StarletteHTTPException)
+        config = ctx.get_app().config
+
         if exc.status_code in {204, 304}:
             return Response(status_code=exc.status_code, headers=exc.headers)
-        return PlainTextResponse(
-            exc.detail, status_code=exc.status_code, headers=exc.headers
+
+        if isinstance(exc.detail, (list, dict)):
+            data = exc.detail
+        else:
+            data = dict(detail=exc.detail, status_code=exc.status_code)
+
+        return config.DEFAULT_JSON_CLASS(
+            data, status_code=exc.status_code, headers=exc.headers
         )
 
 
@@ -44,14 +53,16 @@ class APIExceptionHandler(IExceptionHandler):
     async def catch(
         self, ctx: IExecutionContext, exc: APIException
     ) -> t.Union[Response, t.Any]:
+        assert isinstance(exc, APIException)
+
         config = ctx.get_app().config
-        headers = getattr(exc, "headers", {})
         if isinstance(exc.detail, (list, dict)):
             data = exc.detail
         else:
-            data = {"detail": exc.detail}
+            data = exc.get_details()
+
         return config.DEFAULT_JSON_CLASS(
-            serialize_object(data), status_code=exc.status_code, headers=headers
+            serialize_object(data), status_code=exc.status_code, headers=exc.headers
         )
 
 

ellar/core/exceptions/interfaces.py

@@ -27,7 +27,11 @@ async def catch(
     def __init_subclass__(cls, **kwargs: t.Any) -> None:
         assert (
             cls.exception_type_or_code
-        ), f"exception_type_or_code must be defined. {cls}"
+        ), f"'exception_type_or_code' must be defined. {cls}"
+        if not isinstance(cls.exception_type_or_code, int):
+            assert issubclass(
+                cls.exception_type_or_code, Exception
+            ), "'exception_type_or_code' is not a valid type"
 
 
 class IExceptionMiddlewareService:

ellar/core/guard/base.py

@@ -11,7 +11,9 @@
 
 
 class GuardCanActivate(ABC, metaclass=ABCMeta):
-    exception_class: t.Type[HTTPException] = HTTPException
+    exception_class: t.Union[
+        t.Type[HTTPException], t.Type[APIException]
+    ] = HTTPException
     status_code: int = HTTP_403_FORBIDDEN
     detail: str = "Not authenticated"
 

ellar/core/main.py

@@ -267,10 +267,14 @@ def _finalize_app_initialization(self) -> None:
 
     def add_exception_handler(
         self,
-        exception_handler: IExceptionHandler,
+        *exception_handlers: IExceptionHandler,
     ) -> None:
-        if exception_handler not in self.config.EXCEPTION_HANDLERS:
-            self.config.EXCEPTION_HANDLERS.append(exception_handler)
+        _added_any = False
+        for exception_handler in exception_handlers:
+            if exception_handler not in self.config.EXCEPTION_HANDLERS:
+                self.config.EXCEPTION_HANDLERS.append(exception_handler)
+                _added_any = True
+        if _added_any:
             self.rebuild_middleware_stack()
 
     def rebuild_middleware_stack(self) -> None:

ellar/core/middleware/di.py

@@ -6,6 +6,7 @@
     HTTPConnection as StarletteHTTPConnection,
     Request as StarletteRequest,
 )
+from starlette.responses import JSONResponse
 from starlette.websockets import WebSocket as StarletteWebSocket
 
 from ellar.constants import SCOPE_EXECUTION_CONTEXT_PROVIDER, SCOPE_SERVICE_PROVIDER
@@ -110,3 +111,8 @@ async def error_handler(self, request: Request, exc: Exception) -> Response:
         assert self._500_error_handler
         response = await self._500_error_handler.catch(ctx=execute_context, exc=exc)
         return response
+
+    def error_response(self, request: StarletteRequest, exc: Exception) -> Response:
+        return JSONResponse(
+            dict(detail="Internal server error", status_code=500), status_code=500
+        )