如何在 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 ...

首先，我拥有的每个端点都使用数据库，因此为每个函数添加依赖参数似乎很愚蠢。其次，这似乎是一种迂回的做事方式。我定义了一个全局，然后我定义了一个返回该全局的函数，然后我注入了该函数。我相信有更自然的方式来解决它。

我看到有人建议将我需要的任何内容作为属性添加到应用程序对象

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

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

我错过了什么？

Answer 1

您可以使用应用程序工厂模式来设置您的应用程序。

为避免使用全局或直接向 app 对象添加内容，您可以创建自己的类 Database 来保存连接池。

要将连接池传递给每条路由，您可以使用中间件并将连接池添加到 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()