73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
from fastapi import Request, status
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi.exceptions import RequestValidationError
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
import traceback
|
||
|
||
class BusinessException(Exception):
|
||
"""业务异常基类"""
|
||
def __init__(self, message: str, code: int = 400):
|
||
self.message = message
|
||
self.code = code
|
||
super().__init__(self.message)
|
||
|
||
class DataSourceException(BusinessException):
|
||
"""数据源异常（AkShare等）"""
|
||
def __init__(self, message: str = "数据源异常，请稍后重试"):
|
||
super().__init__(message, code=503)
|
||
|
||
class AuthException(BusinessException):
|
||
"""认证异常"""
|
||
def __init__(self, message: str = "认证失败"):
|
||
super().__init__(message, code=401)
|
||
|
||
class PermissionException(BusinessException):
|
||
"""权限异常"""
|
||
def __init__(self, message: str = "权限不足"):
|
||
super().__init__(message, code=403)
|
||
|
||
async def business_exception_handler(request: Request, exc: BusinessException):
|
||
"""业务异常处理器"""
|
||
return JSONResponse(
|
||
status_code=exc.code,
|
||
content={
|
||
"success": False,
|
||
"error": exc.message,
|
||
"code": exc.code
|
||
}
|
||
)
|
||
|
||
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
||
"""请求参数验证异常处理器"""
|
||
return JSONResponse(
|
||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||
content={
|
||
"success": False,
|
||
"error": "请求参数错误",
|
||
"detail": exc.errors()
|
||
}
|
||
)
|
||
|
||
async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
|
||
"""数据库异常处理器"""
|
||
print(f"数据库错误: {exc}")
|
||
traceback.print_exc()
|
||
return JSONResponse(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
content={
|
||
"success": False,
|
||
"error": "数据库操作失败"
|
||
}
|
||
)
|
||
|
||
async def general_exception_handler(request: Request, exc: Exception):
|
||
"""通用异常处理器"""
|
||
print(f"未捕获异常: {type(exc).__name__}: {exc}")
|
||
traceback.print_exc()
|
||
return JSONResponse(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
content={
|
||
"success": False,
|
||
"error": "服务器内部错误"
|
||
}
|
||
) |