这里我的处理方式是三个文件，分开管理，具体使用视情况而定

自定义异常

error

定义
- error 类通常用于表示 API 响应中的错误信息
- 这些类通常包含状态码、错误消息和时间戳等信息，目的是为了统一和标准化 API 的错误响应格式
使用
- 用于 API 响应时，向客户端返回错误信息
- 适合用于处理请求的结果，尤其是在 RESTful API 中，能够提供一致的错误格式
优点
- 提供了统一的错误响应格式，便于前端处理
- 可以包含额外的元数据（如时间戳），有助于调试和日志记录

from rest_framework import status
from rest_framework.exceptions import APIException


# 自定义异常
class CustomExceptionError(APIException):
    pass


class Success(CustomExceptionError):
    status_code = 200
    default_detail = "请求成功"


class Created(CustomExceptionError):
    status_code = 201
    default_detail = "资源已创建"


class NoContent(CustomExceptionError):
    status_code = 204
    default_detail = "无内容"


class ParamError(CustomExceptionError):
    status_code = 400
    default_detail = "请求无效，可能是参数错误"


class Unauthorized(CustomExceptionError):
    status_code = 401
    default_detail = "未经授权，请重新登录"


class PermissionDenied(CustomExceptionError):
    status_code = 403
    default_detail = "拒绝访问，权限不足"


class ObjectNotFound(CustomExceptionError):
    status_code = 404
    default_detail = "请求的资源不存在"


class ServerError(CustomExceptionError):
    status_code = 500
    default_detail = "服务器内部错误"


class GatewayError(CustomExceptionError):
    status_code = 502
    default_detail = "网关错误"


class ServiceUnavailable(CustomExceptionError):
    status_code = 503
    default_detail = "服务不可用，服务器暂时过载或维护中"


class GatewayTimeout(CustomExceptionError):
    status_code = 504
    default_detail = "网关超时"


class SerializerError(CustomExceptionError):
    status_code = 400
    default_detail = "序列化错误"


class ErrorCode:
    """
    自定义 HTTP 状态码和错误码
    """

    # 认证相关 (10000-10999)
    # UNAUTHORIZED = 10000  # 未登录
    PERMISSION_DENIED = 10001  # 无权限
    TOKEN_EXPIRED = 10002  # Token过期
    TOKEN_INVALID = 10003  # Token无效

    # 参数相关 (40000-40999)
    PARAM_ERROR = 40000  # 参数验证错误
    DATA_NOT_FOUND = 40001  # 未找到数据
    DATA_NOT_VALID = 40002  # 数据错误
    REPEAT_POST = 40003  # 重复提交
    PARAM_MISSING = 40004  # 缺少必要参数
    PARAM_FORMAT_ERROR = 40005  # 参数格式错误

    # 业务相关 (50000-50999)
    BUSINESS_ERROR = 50000  # 业务处理失败
    RESOURCE_NOT_FOUND = 50001  # 资源不存在
    RESOURCE_ALREADY_EXIST = 50002  # 资源已存在
    OPERATION_FAILED = 50003  # 操作失败

    # 系统相关 (60000-60999)
    SYSTEM_ERROR = 60000  # 系统错误
    # SERVICE_UNAVAILABLE = 60001  # 服务不可用
    THIRD_PARTY_ERROR = 60002  # 第三方服务错误
    DATABASE_ERROR = 60003  # 数据库错误
    CACHE_ERROR = 60004  # 缓存错误

    OK = status.HTTP_200_OK  # 请求成功
    CREATED = status.HTTP_201_CREATED  # 资源创建成功
    ACCEPTED = status.HTTP_202_ACCEPTED  # 请求已接受，但尚未处理完成
    NO_CONTENT = status.HTTP_204_NO_CONTENT  # 请求成功但无内容返回
    RESET_CONTENT = status.HTTP_205_RESET_CONTENT  # 请求成功，重置视图内容
    PARTIAL_CONTENT = status.HTTP_206_PARTIAL_CONTENT  # 请求部分内容成功

    # 重定向状态码
    MOVED_PERMANENTLY = status.HTTP_301_MOVED_PERMANENTLY  # 资源已永久转移
    FOUND = status.HTTP_302_FOUND  # 资源已临时转移
    SEE_OTHER = status.HTTP_303_SEE_OTHER  # 请使用 GET 方法获取资源
    NOT_MODIFIED = status.HTTP_304_NOT_MODIFIED  # 资源未被修改
    TEMPORARY_REDIRECT = status.HTTP_307_TEMPORARY_REDIRECT  # 临时重定向
    PERMANENT_REDIRECT = status.HTTP_308_PERMANENT_REDIRECT  # 永久重定向

    # 客户端错误状态码
    BAD_REQUEST = status.HTTP_400_BAD_REQUEST  # 请求格式错误
    UNAUTHORIZED = status.HTTP_401_UNAUTHORIZED  # 未认证
    FORBIDDEN = status.HTTP_403_FORBIDDEN  # 无权限
    NOT_FOUND = status.HTTP_404_NOT_FOUND  # 资源未找到
    METHOD_NOT_ALLOWED = status.HTTP_405_METHOD_NOT_ALLOWED  # 请求方法不被允许
    NOT_ACCEPTABLE = status.HTTP_406_NOT_ACCEPTABLE  # 请求内容不可接受
    REQUEST_TIMEOUT = status.HTTP_408_REQUEST_TIMEOUT  # 请求超时
    CONFLICT = status.HTTP_409_CONFLICT  # 请求冲突，例如重复数据
    GONE = status.HTTP_410_GONE  # 资源永久删除
    LENGTH_REQUIRED = status.HTTP_411_LENGTH_REQUIRED  # 需要指定 Content-Length
    PRECONDITION_FAILED = status.HTTP_412_PRECONDITION_FAILED  # 前提条件未满足
    PAYLOAD_TOO_LARGE = status.HTTP_413_REQUEST_ENTITY_TOO_LARGE  # 请求实体过大
    URI_TOO_LONG = status.HTTP_414_REQUEST_URI_TOO_LONG  # URI 过长
    UNSUPPORTED_MEDIA_TYPE = status.HTTP_415_UNSUPPORTED_MEDIA_TYPE  # 不支持的媒体类型
    TOO_MANY_REQUESTS = status.HTTP_429_TOO_MANY_REQUESTS  # 请求过多

    # 服务器错误状态码
    INTERNAL_SERVER_ERROR = status.HTTP_500_INTERNAL_SERVER_ERROR  # 服务器内部错误
    NOT_IMPLEMENTED = status.HTTP_501_NOT_IMPLEMENTED  # 服务器未实现请求功能
    BAD_GATEWAY = status.HTTP_502_BAD_GATEWAY  # 网关错误
    SERVICE_UNAVAILABLE = status.HTTP_503_SERVICE_UNAVAILABLE  # 服务不可用
    GATEWAY_TIMEOUT = status.HTTP_504_GATEWAY_TIMEOUT  # 网关超时
    HTTP_VERSION_NOT_SUPPORTED = status.HTTP_505_HTTP_VERSION_NOT_SUPPORTED  # 不支持的 HTTP 版本

exception

定义
- exception 类用于表示程序中的异常情况
- 这些类通常是自定义的异常，继承自 Python 的内置异常类（如 Exception 或 APIException），用于在代码中抛出和捕获特定的错误
使用
- 用于在代码中抛出异常，表示某种错误情况（如参数错误、资源未找到等）
- 适合用于业务逻辑中，帮助开发者捕获和处理错误
优点
- 提供了更细粒度的错误处理，能够在代码中明确指出错误的类型
- 使得代码的可读性和可维护性更高，便于调试

from datetime import datetime

from django.http.response import JsonResponse
from rest_framework import status
from rest_framework.exceptions import AuthenticationFailed

from utils.log.logger import logger


class ApiError:
    """API 错误类,用于统一处理API错误响应"""

    DEFAULT_STATUS = 400

    def __init__(self, status=DEFAULT_STATUS, message=None):
        """初始化API错误对象"""
        self.status = status
        self.timestamp = int(datetime.now().timestamp() * 1000)
        self.message = message or "未知错误"  # 提供默认错误信息

    @classmethod
    def error(cls, message):
        """使用默认状态码创建错误对象"""
        return cls(message=message)

    @classmethod
    def error_with_status(cls, status, message):
        """使用自定义状态码创建错误对象"""
        return cls(status=status, message=message)

    def to_dict(self):
        """将错误对象转换为字典格式"""
        return {"status": self.status, "message": self.message, "timestamp": self.timestamp}


class BadCredentialsException(AuthenticationFailed):
    """用户凭证无效异常"""

    def __init__(self, detail="用户名或密码不正确", code=None):
        super().__init__(detail, code)


class BadConfigurationException(RuntimeError):
    """统一关于错误配置信息的异常类"""

    def __init__(self, message=None, *args):
        """构造一个新的运行时异常，允许传入错误信息"""
        super().__init__(message, *args)


class BadRequestException(Exception):
    """统一异常处理"""

    def __init__(self, message: str, status_code: int = status.HTTP_400_BAD_REQUEST):
        self.message = message  # 错误信息
        self.status_code = status_code  # 状态码
        super().__init__(self.message)

    def __str__(self):
        return f"{self.message} (状态码: {self.status_code})"  # 返回错误信息和状态码


class EntityExistException(Exception):
    """实体已存在异常"""

    def __init__(self, entity: str, field: str, value: str):
        """构造一个新的实体已存在异常，包含实体类、字段和对应值"""
        self.message = self.generate_message(entity, field, value)  # 生成异常消息
        super().__init__(self.message)

    @staticmethod
    def generate_message(entity: str, field: str, value: str) -> str:
        """生成异常消息"""
        return f"{entity} 的 {field} '{value}' 已存在"  # 返回实体已存在的消息


class EntityNotFoundException(Exception):
    """实体未找到异常"""

    def __init__(self, entity: str, field: str, value: str):
        """构造一个新的实体未找到异常，包含实体类、字段和对应值"""
        self.message = self.generate_message(entity, field, value)  # 生成异常消息
        super().__init__(self.message)

    @staticmethod
    def generate_message(entity: str, field: str, value: str) -> str:
        """生成异常消息"""
        return f"{entity} 的 {field} '{value}' 不存在"  # 返回实体未找到的消息


class ValidationErrorException(Exception):
    """数据验证错误异常"""

    def __init__(self, message: str):
        """构造一个新的数据验证错误异常"""
        self.message = message  # 错误信息
        super().__init__(self.message)

    def __str__(self):
        return f"验证错误: {self.message}"  # 返回验证错误信息


class UnauthorizedAccessException(Exception):
    """未授权访问异常"""

    def __init__(self, message: str = "未授权访问"):
        self.message = message
        super().__init__(self.message)

    def __str__(self):
        return f"未授权访问: {self.message}"


class ResourceConflictException(Exception):
    """资源冲突异常"""

    def __init__(self, message: str):
        self.message = message
        super().__init__(self.message)

    def __str__(self):
        return f"资源冲突: {self.message}"


class InternalServerErrorException(Exception):
    """内部服务器错误异常"""

    def __init__(self, message: str = "内部服务器错误"):
        self.message = message
        super().__init__(self.message)

    def __str__(self):
        return f"内部服务器错误: {self.message}"


class TimeoutException(Exception):
    """请求超时异常"""

    def __init__(self, message: str = "请求超时"):
        self.message = message
        super().__init__(self.message)

    def __str__(self):
        return f"超时错误: {self.message}"


class ForbiddenException(Exception):
    """禁止访问异常"""

    def __init__(self, message: str = "禁止访问"):
        self.message = message
        super().__init__(self.message)

    def __str__(self):
        return f"禁止访问: {self.message}"


class NotImplementedException(Exception):
    """未实现异常"""

    def __init__(self, message: str = "功能未实现"):
        self.message = message
        super().__init__(self.message)

    def __str__(self):
        return f"未实现错误: {self.message}"


# 全局异常处理类
class GlobalExceptionHandler:
    """全局异常处理类"""

    def _create_json_response(self, api_error):
        """创建统一的 JSON 响应"""
        return JsonResponse(
            {
                "status": api_error.status,
                "message": api_error.message,
                "timestamp": api_error.timestamp,
            },
            status=api_error.status,
        )

    def _log_error(self, error, with_traceback=True):
        """统一的错误日志记录"""
        if with_traceback:
            logger.error(str(error), exc_info=True)
        else:
            logger.error(str(error))

    def handle_exception(self, request, e):
        """处理所有未知异常"""
        self._log_error(e)
        return self._create_json_response(ApiError.error(str(e)))

    def handle_bad_credentials(self, request, e):
        """处理认证凭证异常"""
        message = "用户名或密码不正确" if str(e) == "坏的凭证" else str(e)
        self._log_error(message, with_traceback=False)
        return self._create_json_response(ApiError.error(message))

    def handle_bad_request(self, request, e):
        """处理请求错误异常"""
        self._log_error(e)
        return self._create_json_response(ApiError.error_with_status(400, str(e)))

    def handle_entity_exist(self, request, e):
        """处理实体已存在异常"""
        self._log_error(e)
        return self._create_json_response(ApiError.error(str(e)))

    def handle_entity_not_found(self, request, e):
        """处理实体未找到异常"""
        self._log_error(e)
        return self._create_json_response(ApiError.error_with_status(404, str(e)))

    def handle_validation_error(self, request, e):
        """处理参数验证异常"""
        self._log_error(e)
        message = e.detail[0].get("message") if e.detail else "参数验证失败"
        return self._create_json_response(ApiError.error(message))


handler = GlobalExceptionHandler()

# 自定义异常处理映射
exception_handlers = {
    BadCredentialsException: handler.handle_bad_credentials,
    BadRequestException: handler.handle_bad_request,
    EntityExistException: handler.handle_entity_exist,
    EntityNotFoundException: handler.handle_entity_not_found,
    ValidationErrorException: handler.handle_validation_error,
    UnauthorizedAccessException: handler.handle_exception,
    ResourceConflictException: handler.handle_exception,
    InternalServerErrorException: handler.handle_exception,
    TimeoutException: handler.handle_exception,
    ForbiddenException: handler.handle_exception,
    NotImplementedException: handler.handle_exception,
}

结合使用

在实际开发中，通常建议将这两者结合使用。exception类用于在代码中抛出和捕获错误，而 error 类用于构建 API 响应。这样可以确保在发生错误时，能够通过抛出自定义异常来捕获错误，并通过统一的错误响应格式将错误信息返回给客户端

error类: 适合用于 API 响应，提供统一的错误格式
exception类: 适合用于代码中的错误处理，提供细粒度的错误类型
结合使用这两者，可以提高代码的可读性、可维护性和用户体验

示例
- 当业务逻辑中发生错误时，抛出一个自定义的 BadRequestException
- 在全局异常处理器中捕获这个异常，并使用 ApiError 类构建一个标准化的错误响应

自定义异常处理器

在settings.py 中正确配置自定义异常处理器

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'project.utils.custom_exception_handler',
}

import traceback
from typing import Any, Dict, Optional

from rest_framework.response import Response
from rest_framework.views import exception_handler

from utils.error import ErrorCode, ParamError, CustomExceptionError
from utils.exception import *
from utils.log.logger import logger
from utils.response import pysuper_response


# 带日志记录的异常处理装饰器
def exception_handler_with_logging(func):
    def decorator(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.info(f"---服务器错误: {str(e)}")  # 记录服务器错误信息
            raise ParamError(ErrorCode.PARAM_ERROR)  # 抛出参数错误异常

    return decorator


# 异常拦截器装饰器，用于拦截视图中的异常并记录日志
def view_exception_handler(view_func):
    def decorator(request, *args, **kwargs):
        try:
            return view_func(request, *args, **kwargs)
        except Exception as e:
            logger.error(f"---服务器错误: {str(e)}")
            raise ParamError(ErrorCode.PARAM_ERROR)

    return decorator


# 自定义异常处理器，用于格式化错误响应
def pysuper_ex_handler(exception: Exception, context: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """
    自定义异常处理器，用于格式化错误响应。

    :param exception: 引发的异常
    :param context: 上下文信息
    :return: 格式化的响应数据或 None
    """
    response = exception_handler(exception, context)

    # 如果响应存在，则格式化
    return response and pysuper_response(response)


def custom_exception_handler(exc, context):
    """
    自定义异常处理器
    :param exc: 异常对象
    :param context: 上下文信息
    :return: Response对象
    """
    # 获取DRF默认的异常处理响应
    response = exception_handler(exc, context)

    # TODO：根据实际情况，自定义统一的响应格式
    def format_response(code, message, data=None):
        return Response({"code": code, "message": message, "data": data}, status=code)

    if response is not None:
        # 处理已知异常
        if isinstance(exc, CustomExceptionError):
            return format_response(code=exc.status_code, message=str(exc))

        # 根据异常类型调用对应的处理方法
        for exc_type, handler_method in exception_handlers.items():
            if isinstance(exc, exc_type):
                return handler_method(context["request"], exc)

        # 处理DRF内置异常
        error_code = response.status_code
        error_msg = response.data.get("detail", "未知错误")

        # 定义状态码与错误消息的映射
        status_messages = {
            400: "请求参数错误",
            401: "认证失败",
            403: "权限不足",
            404: "资源不存在",
            405: "请求方法不允许",
            500: "服务器内部错误",
            502: "网关错误",
            503: "服务不可用",
            504: "网关超时",
        }

        if error_code in status_messages:
            error_msg = status_messages[error_code]

        return format_response(
            code=error_code, message=error_msg, data=response.data if hasattr(response, "data") else None
        )

    # 处理未知异常
    logger.error(f"未知异常: {exc}")
    logger.error(traceback.format_exc())
    return format_response(code=500, message=str(exc) or "服务器内部错误")

目录CONTENT

Django--自定义异常处理

自定义异常

error

exception

结合使用

自定义异常处理器

评论区