FastAPI主应用用lifespan时,如何为子应用添加生命周期处理器?
为FastAPI挂载的子应用添加专属生命周期处理函数
问题背景
主应用通过lifespan参数定义了全局生命周期,但无法修改主应用的lifespan生成器,同时add_event_handler方法已废弃且在主应用使用lifespan时无效,需要为挂载的子应用添加独立的启动/关闭逻辑。
问题复现代码
from fastapi import FastAPI # --- 主应用 --- def lifespan(_): print("startup") yield print("shutdown") app = FastAPI(lifespan=lifespan) @app.get("/") async def root(): return {"message": "Hello World"} # --- 子应用 --- sub_app = FastAPI() @sub_app.get("/") async def sub_root(): return {"message": "Hello Sub World"} app.mount("/sub", sub_app) # 以下代码无效,因为主应用使用了lifespan参数 app.on_event("startup")(lambda: print("sub startup")) app.on_event("shutdown")(lambda: print("sub shutdown"))
运行命令:
uvicorn my_app:app --port 8000
解决方案
直接为子应用实例定义专属的lifespan生成器即可。子应用作为独立的FastAPI实例,完全支持lifespan参数,无需依赖主应用的生命周期逻辑。
修改后的完整代码
from fastapi import FastAPI # --- 主应用 --- def lifespan(_): print("startup") yield print("shutdown") app = FastAPI(lifespan=lifespan) @app.get("/") async def root(): return {"message": "Hello World"} # --- 子应用 --- def sub_lifespan(_): print("sub startup") # 在这里添加子应用的启动逻辑,比如初始化资源 yield print("sub shutdown") # 在这里添加子应用的关闭逻辑,比如释放资源 sub_app = FastAPI(lifespan=sub_lifespan) @sub_app.get("/") async def sub_root(): return {"message": "Hello Sub World"} app.mount("/sub", sub_app)
运行效果
启动服务时,控制台会输出:
startup sub startup
关闭服务时,控制台会输出:
sub shutdown shutdown
异步生命周期支持
如果子应用需要执行异步启动/关闭逻辑,只需将sub_lifespan定义为异步生成器:
async def sub_lifespan(_): print("sub async startup") # 执行异步操作,比如异步连接数据库 await some_async_init() yield print("sub async shutdown") # 执行异步清理操作 await some_async_cleanup()
内容的提问来源于stack exchange,提问作者Falko
相关产品推荐
相关产品推荐

