65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError
|
|
from sqlalchemy.orm.exc import StaleDataError
|
|
|
|
from app.api.routes import chat, documents, health, knowledge, sources
|
|
from app.core.config import settings
|
|
from app.db.session import engine
|
|
|
|
|
|
def create_app(db_engine=engine) -> FastAPI:
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
try:
|
|
with db_engine.connect() as connection:
|
|
revision = connection.scalar(text("SELECT version_num FROM alembic_version"))
|
|
if revision != "0002_catalog":
|
|
raise RuntimeError(
|
|
"数据库版本不匹配，请先在 backend 目录执行 alembic upgrade head"
|
|
)
|
|
except ProgrammingError as exc:
|
|
raise RuntimeError(
|
|
"数据库尚未初始化，请先在 backend 目录执行 alembic upgrade head"
|
|
) from exc
|
|
try:
|
|
yield
|
|
finally:
|
|
db_engine.dispose()
|
|
|
|
app = FastAPI(title=settings.app_name, version="0.2.0", lifespan=lifespan)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=settings.cors_origin_list,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
for router in (health.router, chat.router, knowledge.router, sources.router, documents.router):
|
|
app.include_router(router, prefix=settings.api_prefix)
|
|
|
|
@app.exception_handler(StaleDataError)
|
|
async def stale_data_handler(request, exc):
|
|
return JSONResponse(status_code=409, content={"detail": "记录已更新，请刷新后重试"})
|
|
|
|
@app.exception_handler(IntegrityError)
|
|
async def integrity_handler(request, exc):
|
|
return JSONResponse(status_code=409, content={"detail": "数据关联或约束冲突"})
|
|
|
|
@app.exception_handler(OperationalError)
|
|
async def database_unavailable_handler(request, exc):
|
|
return JSONResponse(status_code=503, content={"detail": "数据库暂不可用"})
|
|
|
|
@app.get("/")
|
|
def root():
|
|
return {"message": "Nalati Tourism QA API", "docs": "/docs"}
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|