python - 如何在 FastAPI 中进行持久数据库连接？

Question

我正在用 FastAPI 编写我的第一个项目，但我有点挣扎。特别是，我不确定我应该如何在我的应用程序中使用 asyncpg 连接池。目前我所拥有的是这样的
在 db.py 我有

pgpool = None


async def get_pool():
    global pgpool
    if not pgpool:
        pgpool = await asyncpg.create_pool(dsn='MYDB_DSN')
    return pgpool

然后在单个文件中，我使用 get_pool 作为依赖项。

@router.post("/user/", response_model=models.User, status_code=201)
async def create_user(user: models.UserCreate, pgpool = Depends(get_pool)):
    # ... do things ...

首先，我拥有的每个端点都使用数据库，因此为每个函数添加依赖参数似乎很愚蠢。其次，这似乎是一种迂回的做事方式。我定义一个全局，然后定义一个返回该全局的函数，然后注入(inject)该函数。我相信有更自然的方式来处理它。
我看到人们建议将我需要的任何东西作为属性添加到应用程序对象中

@app.on_event("startup")
async def startup():
    app.pool = await asyncpg.create_pool(dsn='MYDB_DSN')

但是当我有多个带有路由器的文件时它不起作用，我不知道如何从路由器对象访问应用程序对象。
我错过了什么？

Answer 1

您可以使用应用程序工厂模式来设置您的应用程序。
为避免使用全局或直接向应用程序对象添加内容，您可以创建自己的类数据库来保存连接池。
要将连接池传递给每个路由，您可以使用中间件并将池添加到 request.state这是示例代码:

import asyncio

import asyncpg
from fastapi import FastAPI, Request

class Database():

    async def create_pool(self):
        self.pool = await asyncpg.create_pool(dsn='MYDB_DSN')

def create_app():

    app = FastAPI()
    db = Database()

    @app.middleware("http")
    async def db_session_middleware(request: Request, call_next):
        request.state.pgpool = db.pool
        response = await call_next(request)
        return response

    @app.on_event("startup")
    async def startup():
        await db.create_pool()

    @app.on_event("shutdown")
    async def shutdown():
        # cleanup
        pass

    @app.get("/")
    async def hello(request: Request):
        print(request.state.pool)

    return app

app = create_app()