Spaces:
Running
Running
File size: 4,331 Bytes
76b9762 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
"""
异常处理模块，定义应用程序中使用的自定义异常和异常处理器
"""
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from app.log.logger import get_exceptions_logger
logger = get_exceptions_logger()
class APIError(Exception):
"""API错误基类"""
def __init__(self, status_code: int, detail: str, error_code: str = None):
self.status_code = status_code
self.detail = detail
self.error_code = error_code or "api_error"
super().__init__(self.detail)
class AuthenticationError(APIError):
"""认证错误"""
def __init__(self, detail: str = "Authentication failed"):
super().__init__(
status_code=401, detail=detail, error_code="authentication_error"
)
class AuthorizationError(APIError):
"""授权错误"""
def __init__(self, detail: str = "Not authorized to access this resource"):
super().__init__(
status_code=403, detail=detail, error_code="authorization_error"
)
class ResourceNotFoundError(APIError):
"""资源未找到错误"""
def __init__(self, detail: str = "Resource not found"):
super().__init__(
status_code=404, detail=detail, error_code="resource_not_found"
)
class ModelNotSupportedError(APIError):
"""模型不支持错误"""
def __init__(self, model: str):
super().__init__(
status_code=400,
detail=f"Model {model} is not supported",
error_code="model_not_supported",
)
class APIKeyError(APIError):
"""API密钥错误"""
def __init__(self, detail: str = "Invalid or expired API key"):
super().__init__(status_code=401, detail=detail, error_code="api_key_error")
class ServiceUnavailableError(APIError):
"""服务不可用错误"""
def __init__(self, detail: str = "Service temporarily unavailable"):
super().__init__(
status_code=503, detail=detail, error_code="service_unavailable"
)
def setup_exception_handlers(app: FastAPI) -> None:
"""
设置应用程序的异常处理器
Args:
app: FastAPI应用程序实例
"""
@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError):
"""处理API错误"""
logger.error(f"API Error: {exc.detail} (Code: {exc.error_code})")
return JSONResponse(
status_code=exc.status_code,
content={"error": {"code": exc.error_code, "message": exc.detail}},
)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
"""处理HTTP异常"""
logger.error(f"HTTP Exception: {exc.detail} (Status: {exc.status_code})")
return JSONResponse(
status_code=exc.status_code,
content={"error": {"code": "http_error", "message": exc.detail}},
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
request: Request, exc: RequestValidationError
):
"""处理请求验证错误"""
error_details = []
for error in exc.errors():
error_details.append(
{"loc": error["loc"], "msg": error["msg"], "type": error["type"]}
)
logger.error(f"Validation Error: {error_details}")
return JSONResponse(
status_code=422,
content={
"error": {
"code": "validation_error",
"message": "Request validation failed",
"details": error_details,
}
},
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""处理通用异常"""
logger.exception(f"Unhandled Exception: {str(exc)}")
return JSONResponse(
status_code=500,
content={
"error": {
"code": "internal_server_error",
"message": "An unexpected error occurred",
}
},
)
|