mirror of
https://github.com/snailyp/gemini-balance.git
synced 2026-05-07 06:03:11 +08:00
8c62c8121d
- 在Dockerfile中添加默认环境变量配置 - 新增静态资源URL版本化管理功能 - 更新所有模板文件使用static_url函数替代硬编码路径 - 优化错误日志页面移动端按钮布局和响应式设计 - 简化异常处理器返回格式 BREAKING CHANGE: 静态资源URL格式变更，需要重新部署以确保资源正确加载
136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
"""
|
|
异常处理模块，定义应用程序中使用的自定义异常和异常处理器
|
|
"""
|
|
|
|
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=str(exc),
|
|
)
|