152 lines
3.7 KiB
Python
152 lines
3.7 KiB
Python
"""
|
|
API错误处理模块
|
|
"""
|
|
|
|
from typing import Any, Dict, Optional
|
|
from fastapi import HTTPException, status
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.requests import Request
|
|
|
|
|
|
class APIError(HTTPException):
|
|
"""API自定义错误"""
|
|
|
|
def __init__(
|
|
self,
|
|
status_code: int = status.HTTP_400_BAD_REQUEST,
|
|
detail: Any = None,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
error_code: Optional[str] = None,
|
|
):
|
|
super().__init__(status_code=status_code, detail=detail, headers=headers)
|
|
self.error_code = error_code or f"ERR_{status_code}"
|
|
|
|
|
|
class FaceFeatureProcessingError(APIError):
|
|
"""人脸特征处理错误"""
|
|
|
|
def __init__(
|
|
self,
|
|
detail: str = "人脸特征处理失败",
|
|
feature_id: Optional[int] = None,
|
|
):
|
|
if feature_id:
|
|
detail = f"人脸特征处理失败 (特征ID: {feature_id}): {detail}"
|
|
super().__init__(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=detail,
|
|
error_code="FACE_FEATURE_PROCESSING_ERROR"
|
|
)
|
|
|
|
|
|
class FeatureNotFoundError(APIError):
|
|
"""特征记录不存在错误"""
|
|
|
|
def __init__(self, feature_id: int):
|
|
super().__init__(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"特征记录不存在 (ID: {feature_id})",
|
|
error_code="FEATURE_NOT_FOUND"
|
|
)
|
|
|
|
|
|
class DuplicateFeatureError(APIError):
|
|
"""重复特征记录错误"""
|
|
|
|
def __init__(self, person_id: int, feature_type: int):
|
|
super().__init__(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"特征记录已存在 (人员ID: {person_id}, 特征类型: {feature_type})",
|
|
error_code="DUPLICATE_FEATURE"
|
|
)
|
|
|
|
|
|
async def validation_exception_handler(
|
|
request: Request,
|
|
exc: RequestValidationError
|
|
) -> JSONResponse:
|
|
"""
|
|
请求验证异常处理器
|
|
|
|
Args:
|
|
request: 请求对象
|
|
exc: 验证异常
|
|
|
|
Returns:
|
|
JSON响应
|
|
"""
|
|
errors = []
|
|
for error in exc.errors():
|
|
field = ".".join(str(loc) for loc in error["loc"] if loc != "body")
|
|
errors.append({
|
|
"field": field or "body",
|
|
"message": error["msg"],
|
|
"type": error["type"]
|
|
})
|
|
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content={
|
|
"error": {
|
|
"code": "VALIDATION_ERROR",
|
|
"message": "请求参数验证失败",
|
|
"details": errors
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
async def api_error_handler(
|
|
request: Request,
|
|
exc: APIError
|
|
) -> JSONResponse:
|
|
"""
|
|
API错误处理器
|
|
|
|
Args:
|
|
request: 请求对象
|
|
exc: API错误
|
|
|
|
Returns:
|
|
JSON响应
|
|
"""
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={
|
|
"error": {
|
|
"code": exc.error_code,
|
|
"message": exc.detail
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
async def generic_exception_handler(
|
|
request: Request,
|
|
exc: Exception
|
|
) -> JSONResponse:
|
|
"""
|
|
通用异常处理器
|
|
|
|
Args:
|
|
request: 请求对象
|
|
exc: 异常
|
|
|
|
Returns:
|
|
JSON响应
|
|
"""
|
|
# 记录异常到日志
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
logger.error(f"未处理的异常: {exc}", exc_info=True)
|
|
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content={
|
|
"error": {
|
|
"code": "INTERNAL_SERVER_ERROR",
|
|
"message": "服务器内部错误"
|
|
}
|
|
}
|
|
) |