You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在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,可直接mockget_db_adapter的返回值,不需要修改路由逻辑

内容的提问来源于stack exchange,提问作者Conic

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.27 13:15:04