""" FastAPI 应用入口 配置应用实例、中间件、异常处理器等。 """ from contextlib import asynccontextmanager from pathlib import Path from typing import AsyncGenerator from fastapi import FastAPI, Request, status from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from app.api.v1.router import api_router from app.core.config import settings from app.core.exceptions import ( AppException, AuthenticationError, AuthorizationError, ResourceConflictError, ResourceNotFoundError, ValidationError, ) from app.database import close_db, init_db def get_frontend_dir() -> Path: """ 获取前端静态文件目录 支持相对路径和绝对路径配置 """ path = Path(settings.frontend_static_path) if path.is_absolute(): return path # 相对路径基于项目根目录 return Path(__file__).parent.parent / path @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """ 应用生命周期管理 启动时初始化数据库，关闭时释放资源。 """ # 启动时 await init_db() yield # 关闭时 await close_db() def create_application() -> FastAPI: """ 创建 FastAPI 应用实例 Returns: 配置完成的 FastAPI 应用 """ app = FastAPI( title=settings.app_name, version=settings.app_version, description="现代化用户认证系统 API", docs_url="/docs" if not settings.is_production else None, redoc_url="/redoc" if not settings.is_production else None, openapi_url="/openapi.json" if not settings.is_production else None, lifespan=lifespan, ) # 配置 CORS app.add_middleware( CORSMiddleware, allow_origins=["*"] if settings.debug else [], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 注册异常处理器 register_exception_handlers(app) # 注册路由 app.include_router(api_router, prefix=settings.api_v1_prefix) # 健康检查端点 @app.get("/health", tags=["健康检查"]) async def health_check(): """健康检查接口""" return {"status": "healthy", "version": settings.app_version} # 挂载前端静态文件（当静态文件目录存在时） frontend_dir = get_frontend_dir() if frontend_dir.exists(): setup_frontend_routes(app, frontend_dir) return app def setup_frontend_routes(app: FastAPI, frontend_dir: Path) -> None: """ 配置前端静态文件路由 Args: app: FastAPI 应用实例 frontend_dir: 前端静态文件目录 """ # 挂载 _next 静态资源目录（Next.js 构建产物） next_static_dir = frontend_dir / "_next" if next_static_dir.exists(): app.mount( "/_next", StaticFiles(directory=str(next_static_dir)), name="next_static", ) # 挂载 assets 目录（通用静态资源） assets_dir = frontend_dir / "assets" if assets_dir.exists(): app.mount( "/assets", StaticFiles(directory=str(assets_dir)), name="assets", ) # 挂载 static 目录（通用静态资源） static_dir = frontend_dir / "static" if static_dir.exists(): app.mount( "/static", StaticFiles(directory=str(static_dir)), name="static", ) # SPA 路由处理 - 必须在最后注册，以避免覆盖 API 路由 @app.get("/{full_path:path}", include_in_schema=False) async def serve_spa(request: Request, full_path: str): """ SPA 路由处理 - 对于静态文件请求，直接返回文件 - 对于 SPA 路由请求，返回 index.html """ # 跳过 API 路由和健康检查 if full_path.startswith("api/") or full_path == "health": return JSONResponse( status_code=404, content={"success": False, "message": "Not Found"}, ) # 尝试查找静态文件 file_path = frontend_dir / full_path # 如果是目录，尝试查找 index.html if file_path.is_dir(): file_path = file_path / "index.html" # 如果文件存在，返回文件 if file_path.is_file(): return FileResponse( file_path, headers=_get_cache_headers(full_path), ) # 尝试添加 .html 后缀（Next.js 静态导出格式） html_path = frontend_dir / f"{full_path}.html" if html_path.is_file(): return FileResponse(html_path) # 默认返回 index.html（SPA 路由回退） index_path = frontend_dir / "index.html" if index_path.is_file(): return FileResponse(index_path) # 前端文件不存在 return JSONResponse( status_code=404, content={"success": False, "message": "Not Found"}, ) def _get_cache_headers(path: str) -> dict[str, str]: """ 根据文件类型返回适当的缓存头 Args: path: 请求路径 Returns: 缓存相关的 HTTP 头 """ # 静态资源（带 hash 的文件）使用长期缓存 if "/_next/" in path or path.startswith("_next/"): return {"Cache-Control": "public, max-age=31536000, immutable"} # 其他静态文件使用短期缓存 static_extensions = {".js", ".css", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico", ".woff", ".woff2"} if any(path.endswith(ext) for ext in static_extensions): return {"Cache-Control": "public, max-age=86400"} # HTML 文件不缓存 return {"Cache-Control": "no-cache"} def register_exception_handlers(app: FastAPI) -> None: """注册全局异常处理器""" @app.exception_handler(AppException) async def app_exception_handler( request: Request, exc: AppException, ) -> JSONResponse: """应用异常处理""" status_code = status.HTTP_500_INTERNAL_SERVER_ERROR if isinstance(exc, AuthenticationError): status_code = status.HTTP_401_UNAUTHORIZED elif isinstance(exc, AuthorizationError): status_code = status.HTTP_403_FORBIDDEN elif isinstance(exc, ResourceNotFoundError): status_code = status.HTTP_404_NOT_FOUND elif isinstance(exc, ResourceConflictError): status_code = status.HTTP_409_CONFLICT elif isinstance(exc, ValidationError): status_code = status.HTTP_422_UNPROCESSABLE_ENTITY return JSONResponse( status_code=status_code, content={ "success": False, "message": exc.message, "code": exc.code, "details": exc.details, }, ) @app.exception_handler(RequestValidationError) async def validation_exception_handler( request: Request, exc: RequestValidationError, ) -> JSONResponse: """请求验证异常处理""" errors = [] for error in exc.errors(): field = ".".join(str(loc) for loc in error["loc"]) errors.append({ "field": field, "message": error["msg"], "type": error["type"], }) return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={ "success": False, "message": "请求数据验证失败", "code": "VALIDATION_ERROR", "details": {"errors": errors}, }, ) # 创建应用实例 app = create_application()