当前位置：首页 > article >正文

Django后端相应类设计

article 2025/1/11 21:28:48

通用的ApiResponse类：用于生成统一的 API 响应格式。每个响应都包含以下字段（每个接口最终的返回数据格式）：

status_code：HTTP 状态码（如 200、400、500 等）
message：响应的描述信息（如 “Success”、“Resource not found” 等）
data：返回的数据（如果有）
trace_id：请求的唯一标识符

trace_id的作用：

唯一标识请求：每个请求都有一个唯一的 TraceID
日志关联：在日志中记录 TraceID，便于快速定位问题

ApiResponse类核心方法：
1、success 200 请求成功
2、created 201 创建资源成功
3、bad_request 400 后端必要的参数与前端传递的参数不一致。例如缺少必填字段、参数类型错误（如期望是数字，但传入了字符串）
4、unauthorized 401 用户鉴权问题，例如token失效，没有提供有效的身份验证信息
5、forbidden 403 禁止访问，客户端没有权限访问资源
6、not_found 404 请求的资源不存在，URL 路径错误。访问的 API 路径错误等
7、validation_error 422 请求格式正确，但语义错误，服务器无法处理。例如：邮箱格式不正确、密码长度不符合要求。请求数据不符合业务规则
8、conflict 409 资源冲突，请求与当前资源状态冲突.例如：创建资源时，资源已存在。用户重名等
9、internal_server_error 500 服务器内部错误，无法完成请求。例如：数据库连接失败，代码逻辑错误导致的异常
10、service_unavailable 503 服务不可用，服务器暂时无法处理请求。例如：服务器正在维护，第三方 API 服务失效等

中间件middlewares作用：
Django 中的一个全局异常处理器，它的主要作用是捕获 Django 视图函数或中间件中抛出的异常，并根据异常类型返回统一的错误响应
同时，它还支持 TraceID 的生成和传递，便于追踪请求的完整调用链路。

1、中间件会捕获所有未处理的异常（包括 Django 内置异常和自定义异常），根据异常类型，生成对应的错误响应
2、最终会使用 ApiResponse 类生成统一的错误响应格式（status_code、message、data、trace_id）
3、从请求头中获取 TraceID，如果未提供则自动生成一个新的。将异常信息和 TraceID 记录到日志中，便于后续排查问题

整体说明：
1、使用可参考test.py，接口最终返回的数据，需要经过结果处理类ApiResponse，统一进行处理
2、在视图（接口）中，尽量使用ApiResponse类的10个核心方法，捕捉可能发生的异常
3、中间件会捕捉，代码未处理的异常，最终规范为status_code、message、data、trace_id四个字段（中间键与ApiResponse结果处理类是互补的作用）
4、后端返回的数据，最终会规范为status_code、message、data、trace_id四个字段

class StatusCode:
    # 成功状态码
    SUCCESS = 200  # 请求成功
    CREATED = 201  # 创建资源成功

    # 客户端错误（4XX）
    BAD_REQUEST = 400  # 参数错误
    UNAUTHORIZED = 401  # 未授权
    FORBIDDEN = 403  # 禁止访问
    NOT_FOUND = 404  # 资源未找到
    VALIDATION_ERROR = 422  # 参数校验错误
    CONFLICT = 409  # 资源冲突

    # 服务端错误（5XX）
    SERVER_ERROR = 500  # 服务器内部错误
    SERVICE_UNAVAILABLE = 503  # 服务不可用


class ApiException(Exception):
    """
    基础异常类，所有自定义异常继承此类。
    """

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


# 客户端错误（4XX）
class BadRequestException(ApiException):
    """400 Bad Request"""

    def __init__(self, message="Bad request"):
        super().__init__(message, status_code=StatusCode.BAD_REQUEST)


class UnauthorizedException(ApiException):
    """401 Unauthorized"""

    def __init__(self, message="Unauthorized"):
        super().__init__(message, status_code=StatusCode.UNAUTHORIZED)


class ForbiddenException(ApiException):
    """403 Forbidden"""

    def __init__(self, message="Forbidden"):
        super().__init__(message, status_code=StatusCode.FORBIDDEN)


class NotFoundException(ApiException):
    """404 Not Found"""

    def __init__(self, message="Resource not found"):
        super().__init__(message, status_code=StatusCode.NOT_FOUND)


class ValidationErrorException(ApiException):
    """422 Validation Error"""

    def __init__(self, message="Validation error"):
        super().__init__(message, status_code=StatusCode.VALIDATION_ERROR)


class ConflictException(ApiException):
    """409 Conflict"""

    def __init__(self, message="Conflict"):
        super().__init__(message, status_code=StatusCode.CONFLICT)


# 服务端错误（5XX）
class InternalServerErrorException(ApiException):
    """500 Internal Server Error"""

    def __init__(self, message="Internal server error"):
        super().__init__(message, status_code=StatusCode.SERVER_ERROR)


class ServiceUnavailableException(ApiException):
    """503 Service Unavailable"""

    def __init__(self, message="Service unavailable"):
        super().__init__(message, status_code=StatusCode.SERVICE_UNAVAILABLE)

以上为自定义的异常处理类，下述为自定义的中间件

import logging
import uuid

from django.core.exceptions import PermissionDenied
from django.http import Http404

from 响应类设计.exceptions import *
from 响应类设计.response_handler import ApiResponse

logger = logging.getLogger(__name__)


class ExceptionHandlingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 从请求头中获取 Trace ID，如果没有则生成一个新的
        trace_id = request.headers.get('X-Trace-ID', str(uuid.uuid4()))
        request.trace_id = trace_id  # 将 Trace ID 绑定到请求对象

        try:
            response = self.get_response(request)
            return response
        except Exception as e:
            return self.handle_exception(e, trace_id)

    def handle_exception(self, exception, trace_id):
        # 记录异常日志，包含 Trace ID
        logger.error(f"Trace ID: {trace_id}, Error: {str(exception)}", exc_info=True)

        if isinstance(exception, Http404):
            return ApiResponse.not_found(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, PermissionDenied):
            return ApiResponse.forbidden(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, BadRequestException):
            return ApiResponse.bad_request(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, UnauthorizedException):
            return ApiResponse.unauthorized(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, ForbiddenException):
            return ApiResponse.forbidden(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, NotFoundException):
            return ApiResponse.not_found(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, ValidationErrorException):
            return ApiResponse.validation_error(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, ConflictException):
            return ApiResponse.conflict(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, InternalServerErrorException):
            return ApiResponse.internal_server_error(message=str(exception), trace_id=trace_id)
        elif isinstance(exception, ServiceUnavailableException):
            return ApiResponse.service_unavailable(message=str(exception), trace_id=trace_id)
        else:
            # 未知异常，返回 500 状态码
            return ApiResponse.internal_server_error(message="An unexpected error occurred.", trace_id=trace_id)

from django.http import JsonResponse
import uuid

from 响应类设计.exceptions import StatusCode


class ApiResponse:
    """
    通用的 API 响应类，支持 Trace ID 和多种状态码。
    """

    def __init__(self, status_code, message, data=None, trace_id=None):
        self.status_code = status_code
        self.message = message
        self.data = data
        self.trace_id = trace_id or str(uuid.uuid4())  # 生成唯一的 Trace ID

    def to_dict(self):
        """
        将响应数据转换为字典格式。
        """
        return {
            'status_code': self.status_code,
            'message': self.message,
            'data': self.data if self.data is not None else {},
            'trace_id': self.trace_id  # 包含 Trace ID
        }

    def to_json_response(self):
        """
        将响应数据转换为 Django 的 JsonResponse。
        """
        return JsonResponse(self.to_dict(), status=self.status_code)

    # 成功响应
    @staticmethod
    def success(message="Success", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.SUCCESS, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    # 资源创建成功
    @staticmethod
    def created(message="Resource created", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.CREATED, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    # 客户端错误
    @staticmethod
    def bad_request(message="Bad request", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.BAD_REQUEST, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    @staticmethod
    def unauthorized(message="Unauthorized", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.UNAUTHORIZED, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    @staticmethod
    def forbidden(message="Forbidden", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.FORBIDDEN, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    @staticmethod
    def not_found(message="Resource not found", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.NOT_FOUND, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    @staticmethod
    def validation_error(message="Validation error", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.VALIDATION_ERROR, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    @staticmethod
    def conflict(message="Conflict", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.CONFLICT, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    # 服务端错误
    @staticmethod
    def internal_server_error(message="Internal server error", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.SERVER_ERROR, message=message, data=data,
                           trace_id=trace_id).to_json_response()

    @staticmethod
    def service_unavailable(message="Service unavailable", data=None, trace_id=None):
        return ApiResponse(status_code=StatusCode.SERVICE_UNAVAILABLE, message=message, data=data,
                           trace_id=trace_id).to_json_response()

上述为定义的结果处理类，一下为使用说明案例

from django.views import View
import uuid
from .exceptions import BadRequestException, NotFoundException
from .response_handler import ApiResponse


def get_trace_id(request):
    """
    从请求头中获取 Trace ID，如果没有则生成一个新的。
    """
    return request.headers.get('X-Trace-ID', str(uuid.uuid4()))


class MyView(View):
    def get(self, request, *args, **kwargs):
        """
        示例 GET 请求视图，展示如何使用 Trace ID。
        """
        # 获取或生成 Trace ID
        trace_id = get_trace_id(request)

        # 示例：从 URL 参数中获取资源 ID
        resource_id = kwargs.get('resource_id')
        if not resource_id:
            # 如果资源 ID 不存在，返回 400 错误
            raise BadRequestException("Resource ID is required.")

        # 模拟资源查找
        resource = self.get_resource(resource_id)
        if not resource:
            # 如果资源未找到，返回 404 错误
            raise NotFoundException("Resource not found.")

        # 返回成功响应
        return ApiResponse.success(
            message="Resource retrieved successfully",
            data=resource,
            trace_id=trace_id
        )

    def post(self, request, *args, **kwargs):
        """
        示例 POST 请求视图，展示如何使用 Trace ID。
        """
        # 获取或生成 Trace ID
        trace_id = get_trace_id(request)

        # 示例：从请求体中获取数据
        data = request.POST
        if not data.get('name'):
            # 如果名称字段不存在，返回 400 错误
            raise BadRequestException("Name is required.")

        # 模拟创建资源
        resource = self.create_resource(data)
        return ApiResponse.created(
            message="Resource created successfully",
            data=resource,
            trace_id=trace_id
        )

    def get_resource(self, resource_id):
        """
        模拟资源查找逻辑。
        """
        # 这里可以替换为实际的数据库查询逻辑
        if resource_id == "123":
            return {"id": "123", "name": "Example Resource"}
        return None

    def create_resource(self, data):
        """
        模拟资源创建逻辑。
        """
        # 这里可以替换为实际的数据库插入逻辑
        return {"id": str(uuid.uuid4()), "name": data.get('name')}