wren
292b18760c
BusinessException (and its subclass LeauditException) had no FastAPI exception handler, so they escaped as unhandled 500s even when carrying the correct status code (e.g. 403). Add a handler that renders the status code and message as a proper JSON response.
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""FastAPI 应用工厂。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import mimetypes
|
|
import sys
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse, Response
|
|
|
|
from fastapi_admin.config import APP_NAME, APP_CORS_ORIGINS
|
|
from fastapi_admin.config._loader import _find_project_root
|
|
from fastapi_common.fastapi_common_storage.oss_client import OssClient
|
|
from fastapi_common.fastapi_common_web.exception.Base.BusinessException import BusinessException
|
|
|
|
# 确保项目根在 sys.path
|
|
_PROJECT_ROOT = _find_project_root()
|
|
if str(_PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(_PROJECT_ROOT))
|
|
|
|
# fastapi_modules 目录加入路径（使 importlib 能找到各模块）
|
|
_FASTMOD = _PROJECT_ROOT / "fastapi_modules"
|
|
_FASTMOD_LEAUDIT = _FASTMOD / "fastapi_leaudit"
|
|
_NATIVE_LEAUDIT_SRC = _PROJECT_ROOT.parent / "leaudit" / "src"
|
|
if str(_FASTMOD) not in sys.path:
|
|
sys.path.insert(0, str(_FASTMOD))
|
|
if str(_FASTMOD_LEAUDIT) not in sys.path:
|
|
sys.path.insert(0, str(_FASTMOD_LEAUDIT))
|
|
if _NATIVE_LEAUDIT_SRC.exists() and str(_NATIVE_LEAUDIT_SRC) not in sys.path:
|
|
sys.path.insert(0, str(_NATIVE_LEAUDIT_SRC))
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
"""创建并配置 FastAPI 应用。"""
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
import logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logging.getLogger("APP").info(f"{APP_NAME} starting...")
|
|
yield
|
|
logging.getLogger("APP").info(f"{APP_NAME} shutting down...")
|
|
|
|
app = FastAPI(
|
|
title=APP_NAME,
|
|
version="1.0.0",
|
|
lifespan=lifespan,
|
|
docs_url="/api/docs",
|
|
redoc_url=None,
|
|
)
|
|
|
|
# CORS
|
|
origins = [o.strip() for o in APP_CORS_ORIGINS.split(",") if o.strip()]
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# 统一业务异常处理
|
|
@app.exception_handler(BusinessException)
|
|
async def handle_business_exception(request: Request, exc: BusinessException):
|
|
return JSONResponse(
|
|
status_code=exc.status.value,
|
|
content={
|
|
"code": exc.status.value,
|
|
"message": exc.message,
|
|
"data": None,
|
|
},
|
|
)
|
|
|
|
# 注册控制器
|
|
from fastapi_admin.bootstrap_parts.controllers import register_controllers
|
|
register_controllers(app)
|
|
|
|
@app.get("/docauditai/{ObjectPath:path}")
|
|
async def GetCompatObject(ObjectPath: str):
|
|
"""兼容旧前端的对象访问路径，优先读取新桶对象。"""
|
|
Client = OssClient()
|
|
CandidateBuckets = [Client.bucket, "docauditai"]
|
|
for BucketName in CandidateBuckets:
|
|
try:
|
|
if not Client.ObjectExists(ObjectPath, Bucket=BucketName):
|
|
continue
|
|
Content = Client.DownloadBytes(ObjectPath, Bucket=BucketName)
|
|
MediaType = mimetypes.guess_type(ObjectPath)[0] or "application/octet-stream"
|
|
return Response(content=Content, media_type=MediaType)
|
|
except Exception:
|
|
continue
|
|
raise HTTPException(status_code=404, detail="Object not found")
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|