""" 应用入口与工厂 —— 创建 FastAPI 实例，注册生命周期钩子、中间件和路由。 FastAPI application factory for the zm-rag backend. """ from __future__ import annotations from contextlib import asynccontextmanager from typing import Any, AsyncIterator from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from app.config import settings from app.infrastructure.embedding_client import EmbeddingClient from app.infrastructure.es_client import ESClient from app.infrastructure.llm_client import LLMClient from app.infrastructure.mysql_client import MySQLClient from app.infrastructure.neo4j_client import Neo4jClient from app.infrastructure.redis_client import RedisClient from app.utils.logger import configure_logging, get_logger logger = get_logger(__name__) # ── Lifespan ───────────────────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: """应用生命周期钩子：启动时初始化各基础设施客户端并挂载到 app.state，关闭时优雅释放连接。 Startup / shutdown lifecycle hook. * **Startup** – initialise ES, Neo4j and Redis clients; attach to ``app.state``. * **Shutdown** – close all client connections gracefully. """ configure_logging(debug=settings.debug) logger.info( "app_starting", app=settings.app_name, version=settings.app_version, debug=settings.debug, ) # ── Redis（先于 ES 初始化，因为 ESClient 需要 RedisClient 做分布式锁） # Redis first — ESClient needs RedisClient for distributed locking redis_client = RedisClient.from_settings() app.state.redis_client = redis_client logger.info("redis_client_ready", url=settings.redis_url) # ── Elasticsearch（注入 RedisClient 用于 ACL 重算和删除操作的分布式锁） # Inject RedisClient for distributed locking on ACL recompute and delete paths es_client = ESClient.from_settings(redis_client=redis_client) app.state.es_client = es_client logger.info("es_client_ready", host=settings.es_host) # Ensure required indices exist on first run try: await es_client.create_indices() except Exception as exc: # 索引创建失败可能导致后续文档入库和搜索功能不可用，应视为严重错误 # Index creation failure may cause document ingest and search to be unavailable logger.error("es_create_indices_failed", error=str(exc)) # ── Neo4j ──────────────────────────────────────────────────────── neo4j_client = Neo4jClient.from_settings() app.state.neo4j_client = neo4j_client logger.info("neo4j_client_ready", uri=settings.neo4j_uri) # ── MySQL ──────────────────────────────────────────────────────── mysql_client = None if settings.mysql_enabled and settings.mysql_host and settings.mysql_database: try: mysql_client = await MySQLClient.from_settings() await mysql_client.ensure_research_sessions_table() await mysql_client.ensure_convert_log_table() await mysql_client.ensure_research_records_table() await mysql_client.ensure_research_record_reports_table() await mysql_client.ensure_research_record_runs_table() await mysql_client.ensure_notebook_tables() logger.info( "mysql_client_ready", host=settings.mysql_host, database=settings.mysql_database, ) except Exception as exc: logger.warning( "mysql_client_init_failed", error=str(exc), backend=settings.research_session_backend, ) app.state.mysql_client = mysql_client # ── Embedding client ────────────────────────────────────────── embedding_client = EmbeddingClient() app.state.embedding_client = embedding_client logger.info("embedding_client_ready", model=settings.embedding_model) # ── LLM client ──────────────────────────────────────────────── llm_client = LLMClient() app.state.llm_client = llm_client logger.info("llm_client_ready", model=settings.llm_model) # ── Graph admin service (type cache + defaults) ─────────────── # 延迟导入以避免循环依赖 / Lazy import to avoid circular dependency from app.core.graph_admin_service import GraphAdminService graph_admin_service = GraphAdminService(neo4j_client) app.state.graph_admin_service = graph_admin_service try: await graph_admin_service.ensure_default_types() await graph_admin_service.refresh_type_cache() logger.info("graph_admin_service_ready") except Exception as exc: logger.warning("graph_admin_init_failed", error=str(exc)) yield # ── Application is running ──────────────────────────────── # ── Shutdown ───────────────────────────────────────────────────── logger.info("app_shutting_down") await es_client.close() await neo4j_client.close() await redis_client.close() if mysql_client is not None: await mysql_client.close() await embedding_client.close() await llm_client.close() logger.info("app_stopped") # ── App factory ────────────────────────────────────────────────────────────── def create_app() -> FastAPI: """构建并返回配置完成的 FastAPI 应用实例，包括限流中间件、CORS、路由注册和全局异常处理。 Build and return the configured :class:`FastAPI` application.""" app = FastAPI( title=settings.app_name, version=settings.app_version, description="政策法规 RAG 检索增强生成系统", lifespan=lifespan, ) # ── Rate limiting (register before CORS so it runs outermost) ──── if settings.rate_limit_enabled: from app.middleware.rate_limit import RateLimitMiddleware # noqa: E402 app.add_middleware(RateLimitMiddleware) # ── CORS ───────────────────────────────────────────────────────── # 安全修复：限制允许的 HTTP 方法和请求头，避免过于宽松的 CORS 配置 app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["Authorization", "Content-Type", "Accept", "X-Requested-With"], ) # ── Routers ────────────────────────────────────────────────────── from app.api.v1.router import v1_router # noqa: E402 app.include_router(v1_router) # ── Health check ───────────────────────────────────────────────── @app.get("/health", tags=["system"]) async def health_check() -> dict[str, Any]: """Lightweight liveness probe.""" return { "status": "ok", "app": settings.app_name, "version": settings.app_version, } # ── Global error handlers ──────────────────────────────────────── @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse: # 避免拦截 HTTPException，让 FastAPI 内置处理器返回正确的状态码 # Avoid catching HTTPException so FastAPI's built-in handler returns proper status codes if isinstance(exc, HTTPException): raise exc logger.error( "unhandled_exception", path=request.url.path, method=request.method, error=str(exc), exc_info=exc, ) return JSONResponse( status_code=500, content={"detail": "Internal server error"}, ) @app.exception_handler(404) async def not_found_handler(request: Request, exc: Any) -> JSONResponse: return JSONResponse( status_code=404, content={"detail": "Resource not found"}, ) return app # The ASGI entry-point used by ``uvicorn app.main:app``. app = create_app()