通用的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_codeself.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 uuidfrom django.core.exceptions import PermissionDenied
from django.http import Http404from 響應類設計.exceptions import *
from 響應類設計.response_handler import ApiResponselogger = logging.getLogger(__name__)class ExceptionHandlingMiddleware:def __init__(self, get_response):self.get_response = get_responsedef __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 responseexcept Exception as e:return self.handle_exception(e, trace_id)def handle_exception(self, exception, trace_id):# 記錄異常日志,包含 Trace IDlogger.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 uuidfrom 響應類設計.exceptions import StatusCodeclass ApiResponse:"""通用的 API 響應類,支持 Trace ID 和多種狀態碼。"""def __init__(self, status_code, message, data=None, trace_id=None):self.status_code = status_codeself.message = messageself.data = dataself.trace_id = trace_id or str(uuid.uuid4()) # 生成唯一的 Trace IDdef 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)# 成功響應@staticmethoddef 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()# 資源創建成功@staticmethoddef 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()# 客戶端錯誤@staticmethoddef 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()@staticmethoddef 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()@staticmethoddef 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()@staticmethoddef 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()@staticmethoddef 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()@staticmethoddef 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()# 服務端錯誤@staticmethoddef 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()@staticmethoddef 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 ApiResponsedef 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 IDtrace_id = get_trace_id(request)# 示例:從 URL 參數中獲取資源 IDresource_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 IDtrace_id = get_trace_id(request)# 示例:從請求體中獲取數據data = request.POSTif 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 Nonedef create_resource(self, data):"""模擬資源創建邏輯。"""# 這里可以替換為實際的數據庫插入邏輯return {"id": str(uuid.uuid4()), "name": data.get('name')}