Refactored Exception Module, Merged handlers to module and changed a new interface for creating Exception handlers

Author: eadwinCode

ellar/core/exceptions/__init__.py

@@ -0,0 +1,98 @@
+import typing as t
+
+from starlette import status
+from starlette.exceptions import HTTPException, WebSocketException
+
+from .base import APIException, ErrorDetail
+from .validation import RequestValidationError, WebSocketRequestValidationError
+
+__all__ = [
+    "HTTPException",
+    "ImproperConfiguration",
+    "APIException",
+    "WebSocketRequestValidationError",
+    "RequestValidationError",
+    "AuthenticationFailed",
+    "NotAuthenticated",
+    "PermissionDenied",
+    "NotFound",
+    "MethodNotAllowed",
+    "NotAcceptable",
+    "UnsupportedMediaType",
+    "WebSocketException",
+]
+
+
+class ImproperConfiguration(Exception):
+    pass
+
+
+class AuthenticationFailed(APIException):
+    status_code = status.HTTP_401_UNAUTHORIZED
+    default_detail = "Incorrect authentication credentials."
+    default_code = "authentication_failed"
+
+
+class NotAuthenticated(APIException):
+    status_code = status.HTTP_401_UNAUTHORIZED
+    default_detail = "Authentication credentials were not provided."
+    default_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"
+
+
+class NotFound(APIException):
+    status_code = status.HTTP_404_NOT_FOUND
+    default_detail = "Not found."
+    default_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)
+
+
+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)
+
+
+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)

ellar/core/exceptions/base.py

@@ -0,0 +1,128 @@
+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
+
+    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"
+
+    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,
+    ) -> None:
+        if detail is None:
+            detail = self.default_detail
+        if code is None:
+            code = self.default_code
+
+        if status_code is None:
+            status_code = self.status_code
+
+        super(APIException, self).__init__(
+            status_code=status_code,
+            detail=_get_error_details(detail, code),
+            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 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

ellar/core/exceptions/callable_exceptions.py

@@ -0,0 +1,70 @@
+import typing as t
+
+from starlette.concurrency import run_in_threadpool
+from starlette.responses import Response
+
+from ellar.helper import is_async_callable
+
+from ..context import IExecutionContext
+from .interfaces import IExceptionHandler
+
+
+class CallableExceptionHandler(IExceptionHandler):
+    """
+    Default Exception Handler Setup for functions
+
+    usage:
+
+    ```python
+
+    class CustomException(Exception):
+        pass
+
+
+    def exception_handler_fun(ctx: IExecutionContext, exc: Exception):
+        return PlainResponse('Bad Request', status_code=400)
+
+    exception_400_handler = CallableExceptionHandler(
+        exc_class_or_status_code=400, callable_exception_handler= exception_handler_fun
+    )
+    exception_custom_handler = CallableExceptionHandler(
+        exc_class_or_status_code=CustomException, callable_exception_handler= exception_handler_fun
+    )
+
+    # in Config.py
+    EXCEPTION_HANDLERS = [exception_handler, exception_custom_handler]
+
+    ```
+    """
+
+    __slots__ = ("callable_exception_handler", "is_async", "func_args")
+    exception_type_or_code: t.Union[t.Type[Exception], int] = 400
+
+    def __init__(
+        self,
+        *func_args: t.Any,
+        exc_class_or_status_code: t.Union[t.Type[Exception], int],
+        callable_exception_handler: t.Callable[
+            [IExecutionContext, Exception],
+            t.Union[t.Awaitable[Response], Response, t.Any],
+        ]
+    ) -> None:
+        self.callable_exception_handler = callable_exception_handler
+        self.is_async = False
+        self.func_args = func_args
+
+        if not isinstance(exc_class_or_status_code, int):
+            assert issubclass(exc_class_or_status_code, Exception)
+
+        self.exception_type_or_code = exc_class_or_status_code
+
+        if is_async_callable(callable_exception_handler):
+            self.is_async = True
+
+    async def catch(
+        self, ctx: IExecutionContext, exc: Exception
+    ) -> t.Union[Response, t.Any]:
+        args = tuple(list(self.func_args) + [ctx, exc])
+        if self.is_async:
+            return await self.callable_exception_handler(*args)  # type:ignore
+        return await run_in_threadpool(self.callable_exception_handler, *args)

ellar/core/exceptions/handlers.py

@@ -0,0 +1,68 @@
+import typing as t
+
+from starlette import status
+from starlette.exceptions import (
+    HTTPException as StarletteHTTPException,
+    WebSocketException as StarletteWebSocketException,
+)
+from starlette.responses import PlainTextResponse, Response
+
+from ellar.core.context import IExecutionContext
+from ellar.core.exceptions import APIException, RequestValidationError
+from ellar.serializer import serialize_object
+
+from .interfaces import IExceptionHandler
+
+
+class HTTPExceptionHandler(IExceptionHandler):
+    exception_type_or_code = StarletteHTTPException
+
+    async def catch(
+        self, ctx: IExecutionContext, exc: StarletteHTTPException
+    ) -> t.Union[Response, t.Any]:
+        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
+        )
+
+
+class WebSocketExceptionHandler(IExceptionHandler):
+    exception_type_or_code = StarletteWebSocketException
+
+    async def catch(
+        self, ctx: IExecutionContext, exc: StarletteWebSocketException
+    ) -> t.Union[Response, t.Any]:
+        websocket = ctx.switch_to_websocket()
+        await websocket.close(code=exc.code, reason=exc.reason)
+        return None
+
+
+class APIExceptionHandler(IExceptionHandler):
+    exception_type_or_code = APIException
+
+    async def catch(
+        self, ctx: IExecutionContext, exc: APIException
+    ) -> t.Union[Response, t.Any]:
+        config = ctx.get_app().config
+        headers = getattr(exc, "headers", {})
+        if isinstance(exc.detail, (list, dict)):
+            data = exc.detail
+        else:
+            data = {"detail": exc.detail}
+        return config.DEFAULT_JSON_CLASS(
+            serialize_object(data), status_code=exc.status_code, headers=headers
+        )
+
+
+class RequestValidationErrorHandler(IExceptionHandler):
+    exception_type_or_code = RequestValidationError
+
+    async def catch(
+        self, ctx: IExecutionContext, exc: RequestValidationError
+    ) -> t.Union[Response, t.Any]:
+        config = ctx.get_app().config
+        return config.DEFAULT_JSON_CLASS(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            content={"detail": serialize_object(exc.errors())},
+        )