hankin
423baff73b
- Docker bridge 网络隔离（8000 端口封死） - Gunicorn 4 Worker 多进程 - Alembic 数据库迁移基线 - 日志轮转 20m×3 - JWT 密钥 + DB 密码 + CORS 收紧 - 3-2-1 备份链路（NAS + R740-B 冷备） - 连接池 pool_pre_ping + pool_recycle=3600
79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
"""
|
|
自定义异常 + 全局异常处理器
|
|
统一输出格式: { "code": int, "data": Any, "message": str }
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
|
|
# ── 自定义业务异常 ─────────────────────────────────────────
|
|
class BizException(Exception):
|
|
"""通用业务异常，code 对应 HTTP 状态码"""
|
|
|
|
def __init__(self, code: int = 400, message: str = "业务异常", data: Any = None):
|
|
self.code = code
|
|
self.message = message
|
|
self.data = data
|
|
|
|
|
|
class UnauthorizedException(BizException):
|
|
def __init__(self, message: str = "未登录或 Token 已失效"):
|
|
super().__init__(code=401, message=message)
|
|
|
|
|
|
class ForbiddenException(BizException):
|
|
def __init__(self, message: str = "无权访问"):
|
|
super().__init__(code=403, message=message)
|
|
|
|
|
|
class NotFoundException(BizException):
|
|
def __init__(self, message: str = "资源不存在"):
|
|
super().__init__(code=404, message=message)
|
|
|
|
|
|
# ── 统一响应构造 ──────────────────────────────────────────
|
|
def _make_response(code: int, message: str, data: Any = None) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=code,
|
|
content={"code": code, "data": data, "message": message},
|
|
)
|
|
|
|
|
|
# ── 注册全局异常处理器 ────────────────────────────────────
|
|
def register_exception_handlers(app: FastAPI) -> None:
|
|
|
|
@app.exception_handler(BizException)
|
|
async def biz_exception_handler(_req: Request, exc: BizException) -> JSONResponse:
|
|
return _make_response(exc.code, exc.message, exc.data)
|
|
|
|
@app.exception_handler(StarletteHTTPException)
|
|
async def http_exception_handler(
|
|
_req: Request, exc: StarletteHTTPException
|
|
) -> JSONResponse:
|
|
return _make_response(exc.status_code, str(exc.detail))
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(
|
|
_req: Request, exc: RequestValidationError
|
|
) -> JSONResponse:
|
|
# 把 Pydantic 校验错误打平为可读字符串
|
|
errors = []
|
|
for e in exc.errors():
|
|
loc = " -> ".join(str(x) for x in e.get("loc", []))
|
|
errors.append(f"{loc}: {e.get('msg', '')}")
|
|
return _make_response(422, "请求参数校验失败", errors)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def global_exception_handler(
|
|
_req: Request, exc: Exception
|
|
) -> JSONResponse:
|
|
# 兜底：未知异常统一 500
|
|
return _make_response(500, f"服务器内部错误: {exc!s}")
|