FastAPI中数据库客户端存放在哪里？

Question

我有一个 FastAPI 应用程序，通过 Motor 连接 MongoDB。我在应用程序开始时打开 MongoDB 连接，并在结束时使用生命周期关闭它。阅读文档和其他人的 Github 存储库，我发现了 3 种主要的客户端存储方式。

1.使用全局变量进行依赖注入

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase

mongo_client = None

def init_db():
    global mongo_client
    mongo_client = AsyncIOMotorClient("mongo://")

def get_db():
    return mongo_client["mydb"]

def close_db():
    mongo_client.close()
    


@asynccontextmanager
async def lifespan(app):
    init_db()
    yield
    close_db()


app = FastAPI(lifespan=lifespan)

    
@app.get("/")
async def root(db: Annotated[AsyncIOMotorDatabase, Depends(get_db)]):
    return db

但是 Python 中强烈不鼓励使用全局变量。

2.使用单例模式进行依赖注入

基本上和全局变量一样。全局变量的所有优点和缺点。也是全局变量的反模式。

3.以生命周期状态存储数据库

还有另一种方式来存储数据库客户端，它处于生命周期状态，然后它会随着每个请求被标记（如果我理解正确的话），并且可以从那里读取：

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from motor.motor_asyncio import AsyncIOMotorClient


@asynccontextmanager
async def lifespan(app):
    mongo_client = AsyncIOMotorClient("mongo://")
    mongo_db = mongo_client["mydb"]
    yield {"db": mongo_db}
    mongo_client.close()


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def root(request: Request):
    return request.state.db

但是随后我忍不住觉得将数据库客户端传递到请求状态并在每个请求中保持它很奇怪。在请求中存储有关数据库客户端的敏感信息并始终将其与每个请求一起标记似乎是不合适的。

依赖注入似乎更适合这个用例，但它实际上需要全局变量（除非我使用一些 ORM 来为我存储对数据库的引用）。

这里的最佳实践是什么？

Answer 1

在我的项目中，我使用中间件为数据库创建会话，如下所示：

# db_middleware.py

from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import Response
from starlette.requests import Request
from fastapi import FastAPI
from db.session import Session


class DbMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: FastAPI) -> None:
        super().__init__(app)

    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        try:
            request.state.db = Session()
            response = await call_next(request)
        finally:
            if hasattr(request.state, "db") and request.state.db is not None:
                request.state.db.close()
        return response

# main.py

from fastapi import FastAPI
from middlewares.db_middleware import DbMiddleware


def create_app():
    app = FastAPI()
    app.add_middleware(DbMiddleware)
    return app


app = create_app()

根据我的经验，通过这种方法可以控制在特定时间点应使用哪个数据库