Overriding FastAPI dependencies that have parameters

Question

I'm trying to test my FastAPI endpoints by overriding the injected database using the officially recommended method in the FastAPI documentation.

The function I'm injecting the db with is a closure that allows me to build any desired database from a MongoClient by giving it the database name whilst (I assume) still working with FastAPI depends as it returns a closure function's signature. No error is thrown so I think this method is correct:

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

Adding it to the endpoint:

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

And finally, attempting to override it in the tests:

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

However, the dependency doesn't seem to override at all and the test ends up creating a MongoClient with the IP of the production database as in normal execution.

So, any ideas on overriding FastAPI dependencies that are given parameters in their endpoints?

I have tried creating a mock closure function with no success:

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

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

And I have also tried providing the same signature, including the parameter, with still no success:

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

Edit note I'm also aware I can create a separate function that creates my desired database and use that as the dependency, but I would much prefer to use this dynamic version as it's more scalable to using more databases in my apps and avoids me writing essentially repeated functions just so they can be cleanly injected.

Answer 1

I use next fixtures for main db overriding to db for testing:

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

Answer 2

There are two issues with your implementation getting in your way:

1. As you are calling build_db right in the route_receive_notifications function definition, the latter receives nested close function as a dependency. And it's impossible to override it. To fix this you would need to avoid calling your dependency right away and still provide it with db name. For that you can either define a new dependency to inject name into build_db:

# app
def get_db_name():
    return "someDB"

def build_db(name: str = Depends(get_db_name)):
    ...

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

or use functools.partial (shorter but less elegant):

# endpoint
from functools import partial

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

2. FastAPI requires dependency overriding function to have the same signature as the original dependency. Simply switching from *args to a single parameter is enough, although using the same argument name and type makes it easier to support in future. Of course you need to provide the function itself as a value for dependency_overrides without calling it:

def mock_closure(name: str):
    def close():
        return name
    return close

app.dependency_overrides[app.build_db] = mock_closure

Answer 3

My case involved an HTTP client wrapper, instead of a DB. I think it could be applied to your case as well.

Context: I want to inject values for a FastAPI handler's dependency to test various scenarios.

We have a handler with its dependencies

@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)

The function get_client is the dependency I want to override in my tests. It returns a Client object that takes a function that performs an HTTP request to an external service (this function actually wraps aiohttp, but that's not the important part). Here's its barebone definition:

class Client:
  def __init__(request):
    self._request = request
  
  async def request(self, params):
    return await self._request(params)

We want to test various responses from the external service, so we need to build the function that returns a function that returns a Client object (sorry for the tongue-twister), with its params:

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

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

  return get_client()

Then in the various tests we have:

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)
    ...