FastAPI存储数据库连接至app.state报错：AttributeError: 'State'无'db'属性

近期遇到如下错误：

AttributeError: 'State' object has no attribute 'db'

尝试将MongoDB连接存储到app.state.db并在不同路由中使用，但访问失败。相关代码如下：

db.py

from motor.motor_asyncio import AsyncIOMotorClient
from decouple import config

async def get_mongo_connection():
    client = AsyncIOMotorClient(config('MONGO_URL'))
    db = client[config('MONGO_DB')]
    return db

app.py

from utilities.conn.mongo import get_mongo_connection

@app.on_event("startup")
@version(1, 0)
async def startup_event():
    app.state.db = await get_mongo_connection()

routes.py

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi_versioning import VersionedFastAPI, version


router = APIRouter(
    tags=["Account"],
    dependencies=[]
)

@router.post("/send-otp")
@version(1,0)
async def send_otp(request : Request, input : SendOTP):
    print(request.app.state.db)
    return {}

请问如何修复该错误，或是否有更好的方法在所有路由中使用数据库连接？

错误根源 #

你使用fastapi-versioning的VersionedFastAPI包装了原始FastAPI应用，但startup事件是绑定在原始app上的。包装后的新app并不会继承原始app的state属性，导致路由中访问request.app.state.db时找不到对应的属性。

修复方案1：修正startup事件绑定位置 #

如果要继续用app.state存储连接，需要把startup事件绑定到包装后的版本化app上，而非原始app。修改app.py：

from fastapi import FastAPI
from fastapi_versioning import VersionedFastAPI
from utilities.conn.mongo import get_mongo_connection
from routes import router

# 创建原始应用实例
app = FastAPI()
app.include_router(router)

# 生成版本化应用
versioned_app = VersionedFastAPI(app)

# 给版本化应用绑定启动事件
@versioned_app.on_event("startup")
async def startup_event():
    versioned_app.state.db = await get_mongo_connection()

修改后，路由中访问request.app.state.db就能获取到正确的数据库连接。

修复方案2：使用依赖注入（推荐） #

FastAPI官方推荐用依赖注入管理全局资源，这种方式能避免版本化包装带来的state问题，同时代码更清晰、可测试。

步骤1：创建数据库依赖函数 #

在db.py中添加依赖逻辑：

from motor.motor_asyncio import AsyncIOMotorClient
from decouple import config
from fastapi import Depends

# 全局存储客户端实例，避免重复创建连接
_client = None

async def get_mongo_db():
    global _client
    if _client is None:
        _client = AsyncIOMotorClient(config('MONGO_URL'))
    db = _client[config('MONGO_DB')]
    yield db

步骤2：在路由中注入依赖 #

修改routes.py：

from fastapi import APIRouter, Depends
from fastapi_versioning import version
from utilities.conn.mongo import get_mongo_db
from pydantic import BaseModel

# 定义请求模型（若未定义）
class SendOTP(BaseModel):
    # 按需添加字段，例如：
    phone: str

router = APIRouter(
    tags=["Account"],
    dependencies=[]
)

@router.post("/send-otp")
@version(1,0)
async def send_otp(input : SendOTP, db = Depends(get_mongo_db)):
    # 直接使用注入的db实例
    print(db)
    # 示例：查询集合
    # user = await db.users.find_one({"phone": input.phone})
    return {}

可选步骤：处理连接关闭 #

如果需要在应用 shutdown 时关闭MongoDB连接，在app.py中添加事件：

from utilities.conn.mongo import _client

@versioned_app.on_event("shutdown")
async def shutdown_event():
    if _client is not None:
        _client.close()

这种方式的优势：

• 无需依赖request.app.state，彻底规避版本化包装的兼容问题
• 符合FastAPI设计规范，代码可读性、可维护性更强
• 方便在多个路由、模块中复用数据库连接

内容的提问来源于stack exchange，提问作者harsh_jain100