本文目录导读:

在Python中自定义状态码通常有两种场景:一种是HTTP状态码(Web开发),另一种是自定义业务状态码(API设计),下面分别介绍。
自定义HTTP状态码(Web框架)
Flask框架
from flask import Flask, abort, jsonify
app = Flask(__name__)
# 自定义错误处理
@app.errorhandler(409)
def conflict_error(error):
return jsonify({
'code': 409,
'message': '资源冲突',
'details': '请求与当前状态冲突'
}), 409
@app.route('/api/resource', methods=['POST'])
def create_resource():
# 触发自定义状态码
abort(409, description='资源已存在')
FastAPI框架
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
# 自定义异常类
class CustomHTTPException(HTTPException):
def __init__(self, status_code: int, detail: str, error_code: str = None):
super().__init__(status_code=status_code, detail=detail)
self.error_code = error_code
@app.exception_handler(CustomHTTPException)
async def custom_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={
"code": exc.error_code or exc.status_code,
"message": exc.detail,
"path": request.url.path
}
)
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id < 1:
raise CustomHTTPException(
status_code=400,
detail="无效的ID",
error_code="INVALID_ID"
)
return {"item_id": item_id}
自定义业务状态码(API设计)
基础枚举类实现
from enum import Enum
from typing import Dict, Any
class BusinessCode(Enum):
"""业务状态码枚举"""
SUCCESS = (200, "操作成功")
CREATED = (201, "创建成功")
# 客户端错误
BAD_REQUEST = (40000, "请求参数错误")
UNAUTHORIZED = (40100, "未授权访问")
FORBIDDEN = (40300, "禁止访问")
NOT_FOUND = (40400, "资源不存在")
VALIDATION_ERROR = (42200, "参数验证失败")
# 业务逻辑错误
USER_EXISTS = (40901, "用户已存在")
STOCK_INSUFFICIENT = (40902, "库存不足")
ORDER_EXPIRED = (41001, "订单已过期")
# 服务器错误
INTERNAL_ERROR = (50000, "服务器内部错误")
DATABASE_ERROR = (50001, "数据库错误")
def __init__(self, code: int, message: str):
self.code = code
self.message = message
@property
def response(self) -> Dict[str, Any]:
return {
"code": self.code,
"message": self.message,
"status": self.name
}
# 响应封装
class ApiResponse:
@staticmethod
def success(data: Any = None, message: str = "成功"):
return {
"code": 200,
"message": message,
"data": data
}
@staticmethod
def error(business_code: BusinessCode, detail: str = None):
response = business_code.response
if detail:
response["detail"] = detail
return response
# 使用示例
def create_user(username: str):
if len(username) < 3:
return ApiResponse.error(
BusinessCode.VALIDATION_ERROR,
"用户名至少3个字符"
)
if username == "admin":
return ApiResponse.error(
BusinessCode.USER_EXISTS,
"该用户名已被注册"
)
return ApiResponse.success({
"username": username,
"id": 123
})
自定义异常类
class BusinessException(Exception):
"""自定义业务异常"""
def __init__(self, code: int, message: str, status_code: int = 400):
self.code = code
self.message = message
self.status_code = status_code
super().__init__(self.message)
def to_dict(self) -> Dict:
return {
"success": False,
"code": self.code,
"message": self.message
}
class UserNotFoundException(BusinessException):
def __init__(self, user_id: int):
super().__init__(
code=40401,
message=f"用户ID {user_id} 不存在",
status_code=404
)
class InvalidOperationException(BusinessException):
def __init__(self, operation: str):
super().__init__(
code=40001,
message=f"无效操作: {operation}",
status_code=400
)
# 全局异常处理(Flask示例)
@app.errorhandler(BusinessException)
def handle_business_error(error: BusinessException):
return jsonify(error.to_dict()), error.status_code
# 使用
@app.route('/api/user/<int:user_id>')
def get_user(user_id: int):
if user_id <= 0:
raise InvalidOperationException("get_user")
if user_id > 100:
raise UserNotFoundException(user_id)
return jsonify({"user_id": user_id})
自定义HTTP状态码(底层实现)
自定义HTTP响应类
from http.client import HTTPMessage
from http.server import BaseHTTPRequestHandler
class CustomHTTPStatus:
"""自定义HTTP状态码"""
# 信息响应
PROCESSING = 102
# 成功
PARTIAL_CONTENT = 206
# 重定向
MULTIPLE_CHOICES = 300
# 客户端错误
UNPROCESSABLE_ENTITY = 422
LOCKED = 423
FAILED_DEPENDENCY = 424
TOO_EARLY = 425
UPGRADE_REQUIRED = 426
# 服务器错误
NETWORK_AUTHENTICATION_REQUIRED = 511
# 自定义状态码
CUSTOM_ERROR = 520
CUSTOM_TIMEOUT = 524
# 状态码描述
_descriptions = {
102: "处理中",
422: "无法处理的实体",
520: "未知错误",
524: "超时"
}
@classmethod
def get_description(cls, code: int) -> str:
return cls._descriptions.get(code, "未知状态")
# 在基础HTTP服务器中使用
class CustomRequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == '/custom':
self.send_response(CustomHTTPStatus.CUSTOM_ERROR)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(b'{"error": "custom error"}')
最佳实践建议
状态码设计原则
- HTTP状态码:遵循RFC规范,只使用标准状态码
- 业务状态码:使用自定义数字(如4位或5位)
- 前缀规范:
- 2xxx:成功
- 4xxx:客户端错误
- 5xxx:服务器错误
响应格式规范
# 统一的API响应格式
{
"code": 0, # 业务状态码,0表示成功
"message": "success", # 提示信息
"data": {}, # 响应数据
"timestamp": 123456 # 时间戳(可选)
}
状态码对照表示例
STATUS_CODE_MAP = {
# 成功状态码
"SUCCESS": {"code": 200, "http_status": 200},
"CREATED": {"code": 201, "http_status": 201},
# 参数错误
"INVALID_PARAM": {"code": 40000, "http_status": 400},
"MISSING_PARAM": {"code": 40001, "http_status": 400},
# 认证错误
"TOKEN_EXPIRED": {"code": 40100, "http_status": 401},
"INVALID_TOKEN": {"code": 40101, "http_status": 401},
# 业务错误
"USER_EXISTS": {"code": 40900, "http_status": 409},
"ORDER_CLOSED": {"code": 41000, "http_status": 410},
# 系统错误
"DB_ERROR": {"code": 50000, "http_status": 500},
"THIRD_PARTY_ERROR": {"code": 50200, "http_status": 502}
}
选择哪种方式取决于你的应用场景:
- Web应用:使用框架内置的错误处理机制
- API服务:结合枚举类和自定义异常类
- 底层实现:扩展HTTP状态码类
建议统一使用枚举管理状态码,便于维护和文档生成。