FastAPI 中如何动态更新 lifespan 初始化数据而不重启应用

在 fastapi 中，lifespan 事件仅在应用启动/关闭时执行一次，无法直接“更新”其内部变量；但可通过后台异步任务定期刷新共享状态（如 api 响应数据），实现运行时动态更新。

FastAPI 的 @asynccontextmanager 生命周期函数（lifespan）本质是一次性初始化与清理机制——它不提供运行时重入或变量覆盖能力。因此，像 names = requests.get(url).json() 这类初始化逻辑，若写在 yield 之前，只会执行一次；后续即使远程数据变更，names 变量也不会自动同步。

要实现在不重启服务的前提下动态更新 names，核心思路是：将数据存储为可变的共享状态（如全局变量、模块级属性或依赖注入容器中的单例对象），并通过后台常驻异步任务周期性拉取并刷新该状态。

✅ 推荐实践方案如下：

1. 定义可共享的全局状态容器（推荐使用 typing.Dict 或自定义类封装）
2. 使用真正异步 http 客户端（禁用阻塞式 requests）
3. 在 lifespan 中启动后台刷新任务，并确保优雅取消

示例代码（完整可运行）：

import asyncio from fastapi import FastAPI from contextlib import asynccontextmanager import httpx  # ✅ 异步、非阻塞  # 共享状态：所有路由和后台任务均可读写 shared_state = {"names": []}  async def refresh_names_loop():     async with httpx.AsyncClient() as client:         while True:             try:                 response = await client.get("https://api.example.com/names")                 response.raise_for_status()                 shared_state["names"] = response.json()                 print(f"[INFO] names refreshed: {len(shared_state['names'])} items")             except Exception as e:                 print(f"[WARN] Failed to refresh names: {e}")             await asyncio.sleep(3600)  # 每小时刷新一次  @asynccontextmanager async def lifespan(app: FastAPI):     # 启动后台刷新任务（不 await，立即返回）     task = asyncio.create_task(refresh_names_loop())     yield     # 应用关闭时取消任务（避免僵尸协程）     task.cancel()     try:         await task     except asyncio.CancelledError:         pass  app = FastAPI(lifespan=lifespan)  @app.get("/names") def get_names():     return {"data": shared_state["names"]}

⚠️ 关键注意事项：

• 严禁在异步上下文中使用 requests：它会阻塞事件循环，导致整个 FastAPI 应用卡死。务必改用 httpx.AsyncClient 或 aiohttp.Clientsession。
• 状态共享需线程/协程安全：本例中 shared_state 是纯字典赋值（原子操作），适用于简单场景；若需更复杂操作（如增删改查并发访问），建议使用 asyncio.Lock 或迁移至 aiocache 等异步缓存库。
• 任务取消必须显式 await：仅调用 task.cancel() 不足以等待其退出，需 await task 并捕获 CancelledError，确保资源释放。
• 首次加载延迟问题：后台任务启动后首次请求有延迟（如 await asyncio.sleep(3600)）。若需启动即加载，可在 lifespan 的 yield 前主动 await 一次初始化（见进阶优化）。

? 进阶提示：如需支持手动触发刷新（例如通过管理接口），可添加一个 POST /admin/refresh-names 路由，调用 asyncio.create_task(…) 单次刷新，或向后台任务发送信号（如通过 asyncio.Queue 或 asyncio.Event 实现控制通道）。

总结：FastAPI 的 lifespan 不是“可更新的数据源”，而是“生命周期钩子”。真正的动态性来自可变共享状态 + 后台异步任务 + 正确的异步 I/O 工具链。掌握这一模式，你就能轻松实现配置热更新、远程元数据同步、令牌自动续期等生产级需求。