155 lines
4.7 KiB
Python
155 lines
4.7 KiB
Python
"""统一业务异常与错误响应体（需求 §37 → 统一 code+message+detail）。
|
|
|
|
对外格式：{"code": "...", "message": "中文文案", "detail": null}
|
|
未捕获异常兜底为 500 INTERNAL_ERROR，原始异常只进日志。
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import Request, status
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from app.core.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class AppError(Exception):
|
|
"""业务异常基类。"""
|
|
|
|
status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR
|
|
code: str = "INTERNAL_ERROR"
|
|
message: str = "服务内部错误，请稍后重试。"
|
|
|
|
def __init__(
|
|
self,
|
|
message: str | None = None,
|
|
*,
|
|
detail: Any = None,
|
|
code: str | None = None,
|
|
) -> None:
|
|
self.detail = detail
|
|
if message is not None:
|
|
self.message = message
|
|
if code is not None:
|
|
self.code = code
|
|
super().__init__(self.message)
|
|
|
|
def to_response(self) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=self.status_code,
|
|
content={"code": self.code, "message": self.message, "detail": self.detail},
|
|
)
|
|
|
|
|
|
# --- 通用 ---
|
|
|
|
|
|
class NotFoundError(AppError):
|
|
status_code = status.HTTP_404_NOT_FOUND
|
|
code = "NOT_FOUND"
|
|
message = "资源不存在。"
|
|
|
|
|
|
class PermissionDeniedError(AppError):
|
|
status_code = status.HTTP_403_FORBIDDEN
|
|
code = "PERMISSION_DENIED"
|
|
message = "没有权限访问该资源。"
|
|
|
|
|
|
class AuthRequiredError(AppError):
|
|
status_code = status.HTTP_401_UNAUTHORIZED
|
|
code = "AUTH_REQUIRED"
|
|
message = "请先登录。"
|
|
|
|
|
|
class InvalidCredentialsError(AppError):
|
|
status_code = status.HTTP_401_UNAUTHORIZED
|
|
code = "AUTH_INVALID_CREDENTIALS"
|
|
message = "用户名或密码错误。"
|
|
|
|
|
|
class ConflictError(AppError):
|
|
status_code = status.HTTP_409_CONFLICT
|
|
code = "CONFLICT"
|
|
message = "资源冲突。"
|
|
|
|
|
|
class RateLimitedError(AppError):
|
|
status_code = status.HTTP_429_TOO_MANY_REQUESTS
|
|
code = "RATE_LIMITED"
|
|
message = "请求过于频繁，请稍后再试。"
|
|
|
|
|
|
# --- 上传与配额 ---
|
|
|
|
|
|
class StorageQuotaExceededError(AppError):
|
|
status_code = status.HTTP_413_CONTENT_TOO_LARGE
|
|
code = "STORAGE_QUOTA_EXCEEDED"
|
|
message = "存储空间不足。"
|
|
|
|
|
|
class FileTooLargeError(AppError):
|
|
status_code = status.HTTP_413_CONTENT_TOO_LARGE
|
|
code = "FILE_TOO_LARGE"
|
|
message = "单个文件大小超出限制。"
|
|
|
|
|
|
class FileTypeUnsupportedError(AppError):
|
|
status_code = status.HTTP_415_UNSUPPORTED_MEDIA_TYPE
|
|
code = "FILE_TYPE_UNSUPPORTED"
|
|
message = "不支持的文件类型。"
|
|
|
|
|
|
# --- 异常处理器 ---
|
|
|
|
|
|
async def app_error_handler(_: Request, exc: AppError) -> JSONResponse:
|
|
if exc.status_code >= 500:
|
|
logger.error("AppError %s: %s", exc.code, exc.message, exc_info=exc)
|
|
return exc.to_response()
|
|
|
|
|
|
async def validation_error_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content={
|
|
"code": "VALIDATION_ERROR",
|
|
"message": "请求参数不正确。",
|
|
"detail": exc.errors(include_url=False, include_input=False),
|
|
},
|
|
)
|
|
|
|
|
|
_STATUS_CODE_MAP = {
|
|
status.HTTP_404_NOT_FOUND: ("NOT_FOUND", "资源不存在。"),
|
|
status.HTTP_405_METHOD_NOT_ALLOWED: ("METHOD_NOT_ALLOWED", "请求方法不被允许。"),
|
|
status.HTTP_401_UNAUTHORIZED: ("AUTH_REQUIRED", "请先登录。"),
|
|
}
|
|
|
|
|
|
async def http_exception_handler(_: Request, exc: StarletteHTTPException) -> JSONResponse:
|
|
"""把框架层 HTTPException（如未匹配路由的 404）转成统一错误体。"""
|
|
code, message = _STATUS_CODE_MAP.get(exc.status_code, ("HTTP_ERROR", str(exc.detail)))
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={"code": code, "message": message, "detail": None},
|
|
)
|
|
|
|
|
|
async def unhandled_error_handler(request: Request, exc: Exception) -> JSONResponse:
|
|
logger.error("Unhandled: %s %s", request.method, request.url.path, exc_info=exc)
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content={"code": "INTERNAL_ERROR", "message": "服务内部错误，请稍后重试。", "detail": None},
|
|
)
|
|
|
|
|
|
def register_exception_handlers(app: Any) -> None:
|
|
app.add_exception_handler(AppError, app_error_handler)
|
|
app.add_exception_handler(RequestValidationError, validation_error_handler)
|
|
app.add_exception_handler(StarletteHTTPException, http_exception_handler)
|
|
app.add_exception_handler(Exception, unhandled_error_handler) |