Django REST framework 异常处理

# 新建 utils/custom_exception.py

class CustomException(Exception):
    _default_code = 400

    def __init__(
        self,
        message: str = "",
        status_code=status.HTTP_400_BAD_REQUEST,
        data=None,
        code: int = _default_code,
    ):

        self.code = code
        self.status = status_code
        self.message = message
        if data is None:
            self.data = {"detail": message}
        else:
            self.data = data

    def __str__(self):
        return self.message

# utils/custom_exception.py
from rest_framework.views import exception_handler

def custom_exception_handler(exc, context):
    # Call REST framework's default exception handler first,
    # to get the standard error response.
    
    # 这里对自定义的 CustomException 直接返回，保证系统其他异常不受影响
    if isinstance(exc, CustomException):
        return Response(data=exc.data, status=exc.status)
    response = exception_handler(exc, context)
    return response

REST_FRAMEWORK = {
    # ...
    "EXCEPTION_HANDLER": "utils.custom_exception.custom_exception_handler",
}

class ArticleViewSet(viewsets.ModelViewSet):
    """
    允许用户查看或编辑的API路径。
    """
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

    @action(detail=False, methods=["get"], url_name="exception", url_path="exception")
    def exception(self, request, *args, **kwargs):
        # 日志使用 demo
        logger.error("自定义异常")
        raise CustomException(data={"detail": "自定义异常"})

$ curl -H 'Accept: application/json; indent=4' -u admin:admin http://127.0.0.1:8000/api/article/exception/
{
    "detail": "自定义异常"
}

# utils/custom_exception.py

class CustomException(Exception):
    # 自定义code
    default_code = 400
    # 自定义 message
    default_message = None

    def __init__(
            self,
            status_code=status.HTTP_400_BAD_REQUEST,
            code: int = None,
            message: str = None,
            data=None,
    ):
        self.status = status_code
        self.code = self.default_code if code is None else code
        self.message = self.default_message if message is None else message

        if data is None:
            self.data = {"detail": self.message, "code": self.code}
        else:
            self.data = data

    def __str__(self):
        return str(self.code) + self.message

class ExecuteError(CustomException):
    """执行出错"""
    default_code = 500
    default_message = "执行出错"


class UnKnowError(CustomException):
    """执行出错"""
    default_code = 500
    default_message = "未知出错"

class ArticleViewSet(viewsets.ModelViewSet):
    """
    允许用户查看或编辑的API路径。
    """
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

    @action(detail=False, methods=["get"], url_name="exception", url_path="exception")
    def exception(self, request, *args, **kwargs):
        # 日志使用 demo
        logger.error("自定义异常")
        raise CustomException(data={"detail": "自定义异常"})

    @action(detail=False, methods=["get"], url_name="unknown", url_path="unknown")
    def unknown(self, request, *args, **kwargs):
        # 日志使用 demo
        logger.error("未知错误")
        raise UnknownError()

    @action(detail=False, methods=["get"], url_name="execute", url_path="execute")
    def execute(self, request, *args, **kwargs):
        # 日志使用 demo
        logger.error("执行错误")
        raise ExecuteError()

curl -H 'Accept: application/json; indent=4' -u admin:admin http://127.0.0.1:8000/api/article/unknown/
{
    "detail": "未知出错",
    "code": 500
}
$ curl -H 'Accept: application/json; indent=4' -u admin:admin http://127.0.0.1:8000/api/article/execute/
{
    "detail": "执行出错",
    "code": 500
}

def exception_handler(exc, context):
    """
    Returns the response that should be used for any given exception.

    By default we handle the REST framework `APIException`, and also
    Django's built-in `Http404` and `PermissionDenied` exceptions.

    Any unhandled exceptions may return `None`, which will cause a 500 error
    to be raised.
    """
    if isinstance(exc, Http404):
        exc = exceptions.NotFound()
    elif isinstance(exc, PermissionDenied):
        exc = exceptions.PermissionDenied()

    if isinstance(exc, exceptions.APIException):
        headers = {}
        if getattr(exc, 'auth_header', None):
            headers['WWW-Authenticate'] = exc.auth_header
        if getattr(exc, 'wait', None):
            headers['Retry-After'] = '%d' % exc.wait

        if isinstance(exc.detail, (list, dict)):
            data = exc.detail
        else:
            data = {'detail': exc.detail}

        set_rollback()
        return Response(data, status=exc.status_code, headers=headers)

    return None