migrated to DRF exception to support DRF extention libraries

Author: eadwinCode

ninja_extra/controllers/base.py

@@ -22,7 +22,7 @@
 from ninja.security.base import AuthBase
 from ninja.types import DictStrAny
 
-from ninja_extra.exceptions import APIException, NotFound, PermissionDenied
+from ninja_extra.exceptions import APIException, NotFound, PermissionDenied, bad_request
 from ninja_extra.operation import PathView
 from ninja_extra.permissions import AllowAny, BasePermission
 from ninja_extra.shortcuts import (
@@ -69,10 +69,9 @@ def __new__(mcs, name: str, bases: tuple, namespace: dict):
             tag = str(cls.__name__).lower().replace("controller", "")
             cls.tags = [tag]
 
-        if len(bases) > 1:
-            for base_cls in reversed(bases):
-                if issubclass(base_cls, APIController):
-                    compute_api_route_function(base_cls, cls)
+        for base_cls in reversed(bases):
+            if base_cls is not APIController and issubclass(base_cls, APIController):
+                compute_api_route_function(base_cls, cls)
 
         compute_api_route_function(cls)
         if not is_decorated_with_inject(cls.__init__):
@@ -110,6 +109,7 @@ class APIController(ABC, metaclass=APIControllerModelMetaclass):
     Ok = Ok
     Id = Id
     Detail = Detail
+    bad_request = bad_request
 
     @classmethod
     def get_router(cls) -> ControllerRouter:

ninja_extra/exceptions.py

@@ -1,63 +1,254 @@
-from typing import Dict, Optional, Union
+"""
+DRF Exceptions
+"""
+from typing import Any, Dict, List, Optional, Type, Union, no_type_check
 
+from django.http import HttpRequest, HttpResponse, JsonResponse
 from django.utils.encoding import force_str
 from django.utils.translation import gettext_lazy as _
 from ninja.errors import HttpError
 
 from ninja_extra import status
 
 
+@no_type_check
+def _get_error_details(
+    data: Union[List, Dict, "ErrorDetail"],
+    default_code: Optional[Union[str, int]] = None,
+) -> Union[List["ErrorDetail"], "ErrorDetail", Dict[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 = force_str(data)
+    code = getattr(data, "code", default_code)
+    return ErrorDetail(text, code)
+
+
+@no_type_check
+def _get_codes(detail: Union[List, Dict, "ErrorDetail"]) -> Union[str, 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
+
+
+@no_type_check
+def _get_full_details(detail: Union[List, Dict, "ErrorDetail"]) -> 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: Optional[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) -> Any:
+        return hash(str(self))
+
+
 class APIException(HttpError):
     """
-    Subclasses should provide `.status_code` and `.message` properties.
+    Base class for REST framework exceptions.
+    Subclasses should provide `.status_code` and `.default_detail` properties.
     """
 
     status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-    message = _("A server error occurred.")
+    default_detail = _("A server error occurred.")
+    default_code = "error"
 
     def __init__(
         self,
-        message: Union[str, Dict[str, str], None] = None,
-        status_code: Optional[int] = None,
+        detail: Optional[Union[List, Dict, "ErrorDetail", str]] = None,
+        code: Optional[Union[str, int]] = None,
     ) -> None:
-        self.message = message or self.message  # type: ignore
-        self.status_code = status_code or self.status_code
-        super().__init__(status_code=self.status_code, message=self.message)
+        if detail is None:
+            detail = self.default_detail
+        if code is None:
+            code = self.default_code
+
+        self.detail = _get_error_details(detail, code)
 
     def __str__(self) -> str:
-        return self.message
+        return str(self.detail)
+
+    def get_codes(self) -> Union[str, Dict[Any, Any]]:
+        """
+        Return only the code part of the error details.
+
+        Eg. {"name": ["required"]}
+        """
+        return _get_codes(self.detail)  # type: ignore
+
+    def get_full_details(self) -> Dict[Any, Any]:
+        """
+        Return both the message & code parts of the error details.
+
+        Eg. {"name": [{"message": "This field is required.", "code": "required"}]}
+        """
+        return _get_full_details(self.detail)  # type: ignore
+
+
+# The recommended style for using `ValidationError` is to keep it namespaced
+# under `serializers`, in order to minimize potential confusion with Django's
+# built in `ValidationError`. For example:
+#
+# from rest_framework import serializers
+# raise serializers.ValidationError('Value was invalid')
+
+
+class ValidationError(APIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+    default_detail = _("Invalid input.")
+    default_code = "invalid"
+
+    def __init__(
+        self,
+        detail: Optional[Union[List, Dict, "ErrorDetail", str]] = None,
+        code: Optional[Union[str, int]] = None,
+    ):
+        if detail is None:
+            detail = self.default_detail
+        if code is None:
+            code = self.default_code
+
+        # For validation failures, we may collect many errors together,
+        # so the details should always be coerced to a list if not already.
+        if not isinstance(detail, dict) and not isinstance(detail, list):
+            detail = [detail]
+
+        self.detail = _get_error_details(detail, code)
+
+
+class ParseError(APIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+    default_detail = _("Malformed request.")
+    default_code = "parse_error"
 
 
 class AuthenticationFailed(APIException):
     status_code = status.HTTP_401_UNAUTHORIZED
-    message = _("Incorrect authentication credentials.")
+    default_detail = _("Incorrect authentication credentials.")
+    default_code = "authentication_failed"
 
 
 class NotAuthenticated(APIException):
     status_code = status.HTTP_401_UNAUTHORIZED
-    message = _("Authentication credentials were not provided.")
+    default_detail = _("Authentication credentials were not provided.")
+    default_code = "not_authenticated"
 
 
 class PermissionDenied(APIException):
     status_code = status.HTTP_403_FORBIDDEN
-    message = _("You do not have permission to perform this action.")
+    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
-    message = _("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: Optional[str] = None,
-        status_code: Optional[int] = None,
-    ) -> None:
+        detail: Optional[Union[List, Dict, "ErrorDetail", str]] = None,
+        code: Optional[Union[str, int]] = None,
+    ):
         if detail is None:
             detail = force_str(self.default_detail).format(method=method)
-        super().__init__(status_code=status_code, message=detail)
+        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: Optional[Union[List, Dict, "ErrorDetail"]] = None,
+        code: Optional[Union[str, int]] = None,
+        available_renderers: 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: Optional[Union[List, Dict, "ErrorDetail", str]] = None,
+        code: Optional[Union[str, int]] = None,
+    ):
+        if detail is None:
+            detail = force_str(self.default_detail).format(media_type=media_type)
+        super().__init__(detail, code)
+
+
+def server_error(request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
+    """
+    Generic 500 error handler.
+    """
+    data = {"error": "Server Error (500)"}
+    return JsonResponse(data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+def bad_request(
+    request: HttpRequest, exception: Type[APIException], *args: Any, **kwargs: Any
+) -> HttpResponse:
+    """
+    Generic 400 error handler.
+    """
+    data = {"error": "Bad Request (400)"}
+    return JsonResponse(data, status=status.HTTP_400_BAD_REQUEST)

ninja_extra/main.py

@@ -1,17 +1,18 @@
 from importlib import import_module
-from typing import Callable, Optional, Sequence, Type, Union
+from typing import Callable, List, Optional, Sequence, Tuple, Type, Union
 
 from django.core.exceptions import ImproperlyConfigured
 from django.http import HttpRequest, HttpResponse
+from django.urls import URLPattern, URLResolver
 from django.utils.module_loading import module_has_submodule
 from ninja import NinjaAPI
 from ninja.constants import NOT_SET
 from ninja.parser import Parser
 from ninja.renderers import BaseRenderer
 
+from ninja_extra import exceptions
 from ninja_extra.controllers.base import APIController
 from ninja_extra.controllers.router import ControllerRegistry, ControllerRouter
-from ninja_extra.exceptions import APIException
 
 __all__ = [
     "NinjaExtraAPI",
@@ -32,6 +33,7 @@ def __init__(
         auth: Union[Sequence[Callable], Callable, object] = NOT_SET,
         renderer: Optional[BaseRenderer] = None,
         parser: Optional[Parser] = None,
+        app_name: str = "ninja",
     ) -> None:
         super(NinjaExtraAPI, self).__init__(
             title=title,
@@ -45,21 +47,27 @@ def __init__(
             renderer=renderer,
             parser=parser,
         )
+        self.app_name = app_name
+        self.exception_handler(exceptions.APIException)(self.api_exception_handler)
 
-        @self.exception_handler(APIException)
-        def api_exception_handler(
-            request: HttpRequest, exc: APIException
-        ) -> HttpResponse:
-            message = (
-                {"message": exc.message}
-                if not isinstance(exc.message, dict)
-                else exc.message
-            )
-            return self.create_response(
-                request,
-                message,
-                status=exc.status_code,
-            )
+    def api_exception_handler(
+        self, request: HttpRequest, exc: exceptions.APIException
+    ) -> HttpResponse:
+        if isinstance(exc.detail, (list, dict)):
+            data = exc.detail
+        else:
+            data = {"detail": exc.detail}
+
+        return self.create_response(request, data, status=exc.status_code)
+
+    @property
+    def urls(self) -> Tuple[List[Union[URLResolver, URLPattern]], str, str]:
+        _url_tuple = super().urls
+        return (
+            _url_tuple[0],
+            self.app_name,
+            str(_url_tuple[len(_url_tuple) - 1]),
+        )
 
     def register_controllers(self, *controllers: Type[APIController]) -> None:
         for controller in controllers:

ninja_extra/shortcuts.py

@@ -41,7 +41,7 @@ def get_object_or_exception(
             message = "{} with {} was not found".format(
                 queryset.model._meta.object_name, _format_dict(kwargs)
             )
-        raise exception(message=message) from ex
+        raise exception(detail=message) from ex
 
 
 def _format_dict(table: DictStrAny) -> str: