当前位置：首页 > news >正文

FastAPI lifespan示例

news 2025/7/15 16:46:27

在现代 Web 应用开发中，应用的初始化和关闭过程中的资源管理至关重要。特别是在处理需要长时间加载的资源（如大型机器学习模型、数据库连接池、缓存系统等）时，确保这些资源在应用启动时正确加载，并在关闭时优雅释放，是构建健壮服务的关键部分。FastAPI 提供了一个强大且灵活的机制——lifespan 上下文管理器，用于精确控制应用的生命周期事件。

本文通过一个完整的代码示例，深入解析 FastAPI 中 lifespan 的实现原理与使用方式。该示例模拟了典型的后端服务启动流程：加载模型、建立数据库连接、初始化缓存，并在关闭阶段执行对应的清理操作。整个过程通过异步上下文管理器实现，使得资源管理既高效又清晰。同时，通过日志记录和 API 接口暴露当前系统状态，增强了服务的可观测性和可维护性。

接下来将从核心机制、执行流程、运行效果以及优势分析等多个维度详细解读这一实现方案。

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio
import logging# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)# 模拟一个机器学习模型类
class MLModel:def __init__(self):self.model = Noneself.is_loaded = Falseasync def load_model(self):"""模拟加载模型的异步操作"""logger.info("开始加载机器学习模型...")# 模拟耗时的模型加载过程await asyncio.sleep(2)self.model = "预训练模型数据"self.is_loaded = Truelogger.info("模型加载完成！")async def close_model(self):"""模拟关闭模型的清理操作"""logger.info("开始清理模型资源...")await asyncio.sleep(1)self.model = Noneself.is_loaded = Falselogger.info("模型资源清理完成！")def predict(self, data):if not self.is_loaded:raise ValueError("模型尚未加载")return f"预测结果: {data} -> 处理完成"# 全局模型实例
ml_model = MLModel()# 定义lifespan上下文管理器
@asynccontextmanager
async def lifespan(app: FastAPI):"""FastAPI应用的生命周期管理器"""# === 启动阶段 (yield之前) ===logger.info("🚀 FastAPI应用启动中...")try:# 1. 初始化数据库连接logger.info("📊 初始化数据库连接...")await asyncio.sleep(0.5)  # 模拟数据库连接app.state.db_connected = Truelogger.info("✅ 数据库连接成功")# 2. 加载机器学习模型await ml_model.load_model()app.state.model = ml_model# 3. 初始化缓存logger.info("🗄️ 初始化Redis缓存...")await asyncio.sleep(0.3)  # 模拟缓存初始化app.state.cache_ready = Truelogger.info("✅ 缓存初始化完成")# 4. 预热操作logger.info("🔥 执行应用预热...")await asyncio.sleep(0.2)logger.info("✅ 应用预热完成")logger.info("🎉 FastAPI应用启动完成，准备接收请求！")except Exception as e:logger.error(f"❌ 启动过程中发生错误: {e}")raise# === yield: 应用运行阶段 ===yield  # 这里是应用正常运行的时间# === 关闭阶段 (yield之后) ===logger.info("🛑 FastAPI应用开始关闭...")try:# 1. 清理模型资源if hasattr(app.state, 'model'):await app.state.model.close_model()# 2. 关闭数据库连接if hasattr(app.state, 'db_connected'):logger.info("📊 关闭数据库连接...")await asyncio.sleep(0.3)app.state.db_connected = Falselogger.info("✅ 数据库连接已关闭")# 3. 清理缓存if hasattr(app.state, 'cache_ready'):logger.info("🗄️ 清理缓存资源...")await asyncio.sleep(0.2)app.state.cache_ready = Falselogger.info("✅ 缓存资源已清理")logger.info("✅ FastAPI应用关闭完成！")except Exception as e:logger.error(f"❌ 关闭过程中发生错误: {e}")# 创建FastAPI应用实例，传入lifespan参数
app = FastAPI(title="Lifespan示例API",description="演示FastAPI lifespan技术实现原理",lifespan=lifespan  # 关键：将lifespan函数传递给FastAPI
)# API路由
@app.get("/")
async def root():return {"message": "Hello World","db_status": getattr(app.state, 'db_connected', False),"cache_status": getattr(app.state, 'cache_ready', False),"model_status": getattr(app.state, 'model', {}).is_loaded if hasattr(app.state, 'model') else False}@app.get("/predict/{data}")
async def predict(data: str):"""使用预加载的模型进行预测"""try:if not hasattr(app.state, 'model'):return {"error": "模型未初始化"}result = app.state.model.predict(data)return {"input": data, "result": result}except Exception as e:return {"error": str(e)}@app.get("/health")
async def health_check():"""健康检查端点"""return {"status": "healthy","services": {"database": getattr(app.state, 'db_connected', False),"cache": getattr(app.state, 'cache_ready', False),"model": getattr(app.state, 'model', {}).is_loaded if hasattr(app.state, 'model') else False}}if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

技术实现原理详解

1. 核心机制

lifespan 基于 Python 的 asynccontextmanager 装饰器实现，它是一个异步上下文管理器：

yield 之前：应用启动阶段，执行初始化操作
yield 时刻：应用开始接收HTTP请求
yield 之后：应用关闭阶段，执行清理操作

2. 执行流程

启动服务器↓
执行 lifespan 函数 (yield 之前)↓
初始化资源 (数据库、模型、缓存等)↓
到达 yield 点↓
应用开始接收请求 ← 正常运行阶段↓
收到关闭信号 (Ctrl+C 或 SIGTERM)↓
执行 lifespan 函数 (yield 之后)↓
清理资源↓
服务器关闭