如何在FastAPI的多个独立路由器之间共享同一个实例对象
FastAPI 跨路由传递数据库适配器实例的规范方案
绝对不要在两个路由中分别实例化适配器,该操作会导致重复创建数据库连接、浪费资源,甚至可能出现并发连接冲突,以下是两种符合Python规范和FastAPI设计原则的实现方案:
方案1:使用app.state存储全局实例
FastAPI 继承了Starlette的全局状态存储位state,专门用来挂载应用级别的共享对象,实现逻辑最简单:
主应用代码示例
from fastapi import FastAPI from .routers import foo, bar from your_adapter_module import MyDBAdapter # 全局仅实例化一次适配器 adapter = MyDBAdapter() app = FastAPI() # 将适配器挂载到app的state属性中 app.state.db_adapter = adapter app.include_router(foo.router) app.include_router(bar.router) # 可选配置:如果是异步适配器,可通过启停事件处理连接生命周期 @app.on_event("startup") async def init_db_connection(): await adapter.connect() @app.on_event("shutdown") async def close_db_connection(): await adapter.disconnect()
路由侧使用示例
# foo.py 路由文件 from fastapi import APIRouter, Request router = APIRouter() @router.get("/foo/test") async def test_foo(request: Request): # 从请求对象中直接获取全局共享的适配器实例 adapter = request.app.state.db_adapter query_result = adapter.query("SELECT * FROM test_table") return {"data": query_result}
方案2:使用依赖注入(更适合中大型项目)
如果希望代码解耦、方便后续单元测试,可以使用FastAPI原生的依赖注入系统实现调用:
定义公共依赖
在公共模块dependencies.py中编写依赖逻辑:
from your_adapter_module import MyDBAdapter # 全局仅实例化一次适配器 db_adapter = MyDBAdapter() def get_db_adapter(): return db_adapter
路由侧使用示例
# bar.py 路由文件 from fastapi import APIRouter, Depends from your_package.dependencies import get_db_adapter from your_adapter_module import MyDBAdapter router = APIRouter() @router.get("/bar/test") async def test_bar(adapter: MyDBAdapter = Depends(get_db_adapter)): exec_result = adapter.execute("INSERT INTO test_table VALUES (1, 'test')") return {"status": exec_result}
方案选型建议
- 小型项目选方案1,实现成本最低
- 中大型项目、需要做单元测试的项目选方案2,可直接mock
get_db_adapter的返回值,不需要修改路由逻辑
内容的提问来源于stack exchange,提问作者Conic
相关产品推荐
相关产品推荐

