Fastapi中的 lifespan

FastAPI 中的 lifespan 用于管理应用的启动和关闭时的操作，它是一个生命周期管理机制。

核心作用

在应用启动前和关闭后执行一次性的初始化和清理工作。

典型使用场景

1. 资源初始化（启动时）

• 加载机器学习模型
• 建立数据库连接池
• 初始化缓存（Redis）
• 预加载配置文件
• 启动后台任务

2. 资源清理（关闭时）

• 关闭数据库连接
• 清理临时文件
• 保存状态数据
• 关闭外部服务连接

基本用法

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ========== 启动时执行 ==========
    print("应用启动中...")
    # 加载 ML 模型
    ml_models["model"] = load_model("model.pkl")
    # 建立数据库连接
    db_connection = await create_db_pool()
    
    yield  # 应用运行期间
    
    # ========== 关闭时执行 ==========
    print("应用关闭中...")
    # 清理资源
    ml_models.clear()
    await db_connection.close()

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    return {"message": "Hello World"}

实际应用示例

示例 1: 加载机器学习模型

from contextlib import asynccontextmanager
from fastapi import FastAPI

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时加载模型（耗时操作）
    ml_models["classifier"] = load_heavy_model()
    yield
    # 关闭时清理
    ml_models.clear()

app = FastAPI(lifespan=lifespan)

@app.post("/predict")
async def predict(data: dict):
    model = ml_models["classifier"]
    return {"prediction": model.predict(data)}

示例 2: 数据库连接管理

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时建立连接池
    app.state.db = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/db"
    )
    yield
    # 关闭时释放连接
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)

示例 3: 后台任务

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动后台任务
    task = asyncio.create_task(periodic_cleanup())
    yield
    # 取消后台任务
    task.cancel()
    await task

执行时机

1. uvicorn main:app 启动命令
         ↓
2. 执行 lifespan 中 yield 之前的代码
         ↓
3. 应用开始接受请求
         ↓
4. Ctrl+C 或收到关闭信号
         ↓
5. 执行 lifespan 中 yield 之后的代码
         ↓
6. 应用完全关闭

为什么需要 lifespan？

不使用 lifespan 的问题：

#  每次请求都加载模型（慢！）
@app.get("/predict")
async def predict():
    model = load_model()  # 每次都加载
    return model.predict()

使用 lifespan：

#  只在启动时加载一次
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.model = load_model()  # 只加载一次
    yield

@app.get("/predict")
async def predict(request: Request):
    return request.app.state.model.predict()

旧版本的等价方式

FastAPI 早期版本使用事件处理器：

# 旧方式（已弃用）
@app.on_event("startup")
async def startup():
    print("启动")

@app.on_event("shutdown")
async def shutdown():
    print("关闭")

# 新方式（推荐）
@asynccontextmanager
async def lifespan(app: FastAPI):
    print("启动")
    yield
    print("关闭")

app = FastAPI(lifespan=lifespan)

总结

lifespan 能够：

• 在应用启动时执行耗时的初始化操作
• 确保资源在应用关闭时被正确清理
• 避免在每次请求时重复初始化
• 更好地管理应用状态和资源

这对于生产环境中的性能优化和资源管理非常重要。