覆盖具有参数的 FastAPI 依赖项

Joe*_*oon 7 python dependency-injection pytest fastapi

我正在尝试使用FastAPI 文档中官方推荐的方法覆盖注入的数据库来测试我的 FastAPI 端点。

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

# app
def build_db(name: str):
    def close():
          return build_singleton_whatever(MongoClient, args....)
     return close
Run Code Online (Sandbox Code Playgroud)

将其添加到端点:

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

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

# pytest
# test_endpoint.py
fastapi_app.dependency_overrides[app.build_db] = lambda x: lambda: x
Run Code Online (Sandbox Code Playgroud)

然而,依赖关系似乎根本没有被覆盖,测试最终会像正常执行一样使用生产数据库的 IP 创建一个 MongoClient。

那么,关于重写在其端点中给定参数的 FastAPI 依赖项有什么想法吗?

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

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

app.dependency_overrides[app.build_db] = mock_closure('otherDB')
Run Code Online (Sandbox Code Playgroud)

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

app.dependency_overrides[app.build_db('someDB')] = mock_closure('otherDB')
Run Code Online (Sandbox Code Playgroud)

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

gRi*_*yGR 1

我的案例涉及 HTTP 客户端包装器,而不是数据库。我认为它也可以适用于您的情况。

上下文:我想为 FastAPI 处理程序的依赖项注入值来测试各种场景。

我们有一个处理程序及其依赖项

@router.get("/{foo}")
async def get(foo, client = Depends(get_client)): # get_client is the key to override
  client = get_client()
  return await client.request(foo)
Run Code Online (Sandbox Code Playgroud)

该函数get_client是我想在测试中覆盖的依赖项。它返回一个Client对象,该对象接受一个对外部服务执行 HTTP 请求的函数(这个函数实际上包装了aiohttp,但这不是重要的部分)。这是它的准系统定义:

class Client:
  def __init__(request):
    self._request = request
  
  async def request(self, params):
    return await self._request(params)
Run Code Online (Sandbox Code Playgroud)

我们想要测试来自外部服务的各种响应,因此我们需要构建一个函数,该函数返回一个返回Client对象的函数(抱歉绕口令),其参数如下:

def get_client_getter(response):
  async def request_mock(*args, **kwargs):
    return response

  def get_client():
    return Client(request=request_mock)

  return get_client()
Run Code Online (Sandbox Code Playgroud)

然后在各种测试中我们有:

def test_1():
    app.dependency_overrides[get_client] = get_client_getter(1)
    ...

def test_true():
    app.dependency_overrides[get_client] = get_client_getter(True)
    ...

def test_none():
    app.dependency_overrides[get_client] = get_client_getter(None)
    ...
Run Code Online (Sandbox Code Playgroud)