python - 覆盖具有参数的 FastAPI 依赖项

Question

我正在尝试通过使用 FastAPI documentation 中官方推荐的方法覆盖注入(inject)的数据库来测试我的 FastAPI 端点.

我注入(inject)数据库的函数是一个闭包，它允许我通过给它数据库名称从 MongoClient 构建任何所需的数据库，同时(我假设)仍然使用 FastAPI depends 因为它返回闭包函数的签名。没有抛出任何错误，所以我认为这个方法是正确的:

# app
def build_db(name: str):
    def close():
          return build_singleton_whatever(MongoClient, args....)
     return close

将其添加到端点:

# endpoint
@app.post("/notification/feed")
async def route_receive_notifications(db: Database = Depends(build_db("someDB"))):
   ...

最后，尝试在测试中覆盖它:

# pytest
# test_endpoint.py
fastapi_app.dependency_overrides[app.build_db] = lambda x: lambda: x

但是，依赖项似乎根本没有覆盖，测试最终会像正常执行一样使用生产数据库的 IP 创建一个 MongoClient。

那么，关于覆盖在端点中给定参数的 FastAPI 依赖项有什么想法吗？

我尝试创建一个模拟闭包函数但没有成功:

def mock_closure(*args):
    def close():
        return args
    return close

app.dependency_overrides[app.build_db] = mock_closure('otherDB')

我也尝试提供相同的签名，包括参数，但仍然没有成功:

app.dependency_overrides[app.build_db('someDB')] = mock_closure('otherDB')

编辑笔记 我也知道我可以创建一个单独的函数来创建我想要的数据库并将其用作依赖项，但我更愿意使用这个动态版本，因为它更具可扩展性在我的应用程序中使用更多数据库，避免我编写本质上重复的函数，以便可以干净地注入(inject)它们。

Answer 1

我使用下一个 fixtures 将主数据库覆盖到数据库进行测试:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

from settings import get_settings


@pytest.fixture()
async def get_engine():
    engine = create_async_engine(get_settings().test_db_url)
    yield engine
    await engine.dispose()


@pytest.fixture()
async def db_session(get_engine) -> AsyncSession:
    async with get_engine.begin() as connection:
        async with async_session(bind=connection) as session:
            yield session
            await session.close()


@pytest.fixture()
def override_get_async_session(db_session: AsyncSession) -> Callable:
    async def _override_get_async_session():
        yield db_session

    return _override_get_async_session