main.py
2.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
"""FastAPI应用入口."""
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.router import api_router
from app.core.exceptions import EmbeddingServiceError
from app.core.config import settings
from app.db import connect as db_client_connect, disconnect as db_client_disconnect
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
"""应用生命周期管理."""
# 启动时建立数据库连接
try:
await db_client_connect()
db_type = settings.get("db_type", "es")
print(f"✓ {db_type.upper()} database connected")
except Exception as e:
print(f"✗ Failed to connect to database: {e}")
raise
yield
# 关闭时断开数据库连接
try:
await db_client_disconnect()
db_type = settings.get("db_type", "es")
print(f"✓ {db_type.upper()} database disconnected")
except Exception as e:
print(f"✗ Error disconnecting from database: {e}")
def create_app() -> FastAPI:
"""创建FastAPI应用."""
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="基于向量的CRUD服务，支持多种Embedding场景",
lifespan=lifespan,
docs_url="/docs",
redoc_url="/redoc",
)
# 配置CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 注册路由
app.include_router(api_router)
# 异常处理
@app.exception_handler(EmbeddingServiceError)
async def embedding_service_exception_handler(request, exc):
from fastapi.responses import JSONResponse
return JSONResponse(
status_code=500,
content={"code": exc.code, "message": exc.message},
)
# @app.get("/health", tags=["健康检查"])
# async def health_check():
# """健康检查端点."""
# db_healthy = await db_client.health_check()
# db_type = settings.get("db_type", "es")
# return {
# "status": "healthy" if db_healthy else "unhealthy",
# "database": {
# "type": db_type,
# "status": "connected" if db_healthy else "disconnected",
# },
# }
@app.get("/", tags=["根路径"])
async def root():
"""根路径."""
return {
"name": settings.app_name,
"version": settings.app_version,
"docs": "/docs",
}
return app
# 创建应用实例
app = create_app()
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main:app",
host='0.0.0.0',
port=settings.port,
reload=settings.debug,
)