[Q] Provider override not working with subcontainers

[TojikCZ]

Sorry for the second post today, i am kinda at the end of my ropes here
I would be so happy if someone manages to get me unstuck.

I have three levels of containers. Root, Main and module specific
In component/unit tests when I want to override a provider, it only works if I instantiate the module specific container like UtilsContainer and orverride its database provider
If I instead make an instance of RootContainer and do
with container.main_package.utils_package.database.override(mocked_db):
it does not work and the original container is used

I placed a raise statement before my RootContainer init in the main code. Does not get triggered, so that instance does not get created from a spurious import somewhere

I am trying to do this without getting a reference to the root container, instead hoping that I can create a new instance in my test, override there and have that injected in whatever I call from the test

My service does not get the container reference, it injects like so: database: PostgresDatabase = Provide[UtilsContainer.database]

conftest.py

from unittest.mock import AsyncMock

import pytest

from somewhere.adapters.dependency_injection.root_container import (
    RootContainer,
)
from somewhere.adapters.postgres import PostgresDatabase
from somewhere.utilities.schemas import Health


@pytest.fixture
def container(request):
    """Creates container"""
    container = RootContainer()
    container.init_resources()

    overrides = request.param
    generators = []
    for override_fn in overrides:
        generators.append(override_fn(container))

    for generator in generators:
        next(generator)

    yield container

    for generator in generators:
        try:
            next(generator)
        except StopIteration:
            pass

    container.unwire()


def unhealthy_database(container):
    mocked_db = AsyncMock(spec=PostgresDatabase)
    mocked_db.health.return_value = Health.FAILING
    with container.main_package.utils_package.database.override(mocked_db):
        yield

test_utilities.py

import pytest

from somewhere.utilities.schemas import Health
from somewhere.utilities.service import get_health_status
from tests.component.conftest import unhealthy_database, healthy_database


@pytest.mark.asyncio
@pytest.mark.parametrize("container", [[unhealthy_database]], indirect=True)
async def test_unhealthy(container):
    health_status = await get_health_status()
    assert health_status.app == Health.OK
    assert health_status.db == Health.FAILING

I am open to distilling this into a minimal example, but I fear I am missing something simple.
Thank you

[ZipFile]

Nested containers is an area we can improve. In your particular case, proper incantation would be database: PostgresDatabase = Provide[RootContainer.main_package.utils_package.database]. But this is sub-optimal since it may lead to recursive imports, so in this case database: PostgresDatabase = Provide["main_package.utils_package.database"] would be the way to go. Which is not super optimal either, since it makes it difficult to debug typos/renemeals/etc....

I've adapted your example to be a standalone runnable test, it should work as you expect:

test_di_issue851.py

from dataclasses import dataclass
from enum import StrEnum, auto
from unittest.mock import AsyncMock

import pytest

from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject


class Health(StrEnum):
    OK = auto()
    FAILING = auto()


@dataclass
class HealthCheckResponse:
    app: Health
    db: Health


class PostgresDatabase:
    async def health(self) -> Health:
        return Health.OK


class UtilsContainer(containers.DeclarativeContainer):
    database = providers.Factory(PostgresDatabase)


class MainContainer(containers.DeclarativeContainer):
    utils_package = providers.Container(UtilsContainer)


class RootContainer(containers.DeclarativeContainer):
    wiring_config = containers.WiringConfiguration(modules=[__name__])
    main_package = providers.Container(MainContainer)


@inject
async def get_health_status(
    database: PostgresDatabase = Provide["main_package.utils_package.database"],
) -> HealthCheckResponse:
    return HealthCheckResponse(
        app=Health.OK,
        db=await database.health(),
    )


@pytest.fixture
def container(request):
    """Creates container"""
    container = RootContainer()
    container.init_resources()

    overrides = request.param
    generators = []
    for override_fn in overrides:
        generators.append(override_fn(container))

    for generator in generators:
        next(generator)

    yield container

    for generator in generators:
        try:
            next(generator)
        except StopIteration:
            pass

    container.unwire()


def unhealthy_database(container):
    mocked_db = AsyncMock(spec=PostgresDatabase)
    mocked_db.health.return_value = Health.FAILING
    with container.main_package.utils_package.database.override(mocked_db):
        yield


@pytest.mark.asyncio
@pytest.mark.parametrize("container", [[unhealthy_database]], indirect=True)
async def test_unhealthy(container):
    health_status = await get_health_status()
    assert health_status.app == Health.OK
    assert health_status.db == Health.FAILING

[TojikCZ]

Thank you so much for this.
The key piece of info I am taking away here is that I want to specify the full path to the provider. In case of the string representation minus the root container itself.

I would like to wrap my brain around how all this works. The doc is excellent in going over most usage scenarios, but I feel like it lacks in the deep end territory. Do you happen to have a resource to point me towards, or is reading code the only way for now? We are trying to use this as our DI "framework" at work and it will get hard to advocate for it if nobody knows how it works.

Again, thank you a ton for answering both of my posts.

[ZipFile]

reading code the only way for now?

Unfortunately, this. Also it is worth checking the issues page, people did figure out some common use cases not outlined in the docs.

The good news - past few weeks I've been working on updating docs/examples based on few recent changes + my experience using DI at scale. At my current place we have 50+ Python microservices of various sizes, majority of them use DI. The reason I've volunteered to assist @rmk135 in maintenance to begin with, as its usage is 100% my initiative.

The bad news is - I have a limited time to work on it too.

Again, thank you a ton for answering both of my posts.

NP, feel free to ask more if needed. I'll consolidate the knowledge eventually.

[TojikCZ]

Thanks, I think I'll take you up on that. Sorry 😅

You mentioned using @inject as a last resort here , whereas I felt like that is the default way to use DI.
But maybe it was meant just as a general rule - don't use DI for everything, only the IO stuff like DB, sentry, logging and whatnot and I agree with that.

And one that's on topic - The example works, but only if i wire the root container to the module where I use it.

So I think I have the pattern. In tests I need to instantiate and override using the exact path that got used to wire the container to the modules I'm testing.
I cannot instantiate the root container and override through the full dot path to the provider, because the wire call on the subcontainer that wired the module in the first place was not aware of the full path anymore, so my override does not match.
Is this close enough?

[ZipFile]

I felt like that is the default way to use DI.

Wiring is a monkey-patch level of sorcery, which you rarely want in your code (for sanity reasons). Think of DI as a glue to assemble all the components together. By adding explicit @inject + Provide[...] your code now aware of the implementation detail how everything is assembled, which violates Separation of Concerns idea. If you ever move your db stuff away from main_package.utils_package.database think of collateral damage it'll inflict.

So the basic pattern is to go full java-style constructor injection. I.e. provide every dependency through the __init__, which is what DeclarativeContainer is designed to do. And only @inject where it is not feasible to fight the framework (FastAPI/Django views, Celery tasks, etc...).

As a bonus of such approach, you:

• Can do unit testing without DI involvement/interference at all. For unit testing, Pytest fixtures are much more flexible dependency injection solution either way. Note: keyword here is unit testing, for other kind of tests you pretty much want DI container around.
• If you ever decide to move away to another DI framework, it wont be such a huge pain in the ass.

And one that's on topic - The example works, but only if i wire the root container to the module where I use it.

If I got you right, this is what we configured it to do so. modules argument accepts list of modules to check for @injects, packages if you want to do it recursively. See docs for details.

I cannot instantiate the root container and override through the full dot path to the provider, because the wire call on the subcontainer that wired the module in the first place was not aware of the full path anymore, so my override does not match.

As I've mentioned before, nesting containers could do better. In my projects we never call wire on anything but the root container. So paths are always absolute.

[TojikCZ]

Well, your approach - having even your services wired up with their dependencies in the Container in my mind now makes your API layer dependent on injects, or on having a reference to the container instance. For now, I am opting to test out the opposite way to your approach. I am going to inject like this
database: PostgresDatabase = Provide["database"]
Wiring the module container to the service instead of the RootContainer
I'm going to override like this [simplified]
UtilsContainer().database.override(Mock())
This way I hope my services do not care about the container structure and unless I am testing interaction between components, the unit test setup seems manageable for now.

You'll probably know soon enough when this approach crashes and burns. 😅

[ZipFile]

Well, your approach - having even your services wired up with their dependencies in the Container in my mind now makes your API layer dependent on injects, or on having a reference to the container instance.

You're absolutely right, but there is a catch. Our views looks like this:

views.py

@router.get("/templates/{id}/preview/")
@inject
async def template_preview(
    id: int,
    user: User = Depends(user),
    query: PreviewTemplateQuery = Depends(Provide["preview_template_query"]),
) -> ResponseData[TemplatePreview]:
    preview = await query.query(
        template_id=id,
        company_id=user.company_id,
    )
    return single_item_response(preview)

We're trying to practice CQRS, so "command" looks not so different from example above (queries). Our views have almost no logic, aside from schema validation (which is done mostly by Pydantic already). I cannot share the specifics, but we were heavily inspired by the Cosmic Python book, I recommend you checking it too. It's quite difficult to do it right the first time, you might want to experiment with something simpler like Clean Architecture (in the example above you'll be injecting a "use case" instead of query or command).

We nest our containers through Container provider, btw. There is no proper documentation page for it, but there are few miniapp examples. Maybe this will help.

You'll probably know soon enough when this approach crashes and burns. 😅

I've seen so much AttributeError: 'Provide' object has no attribute ... errors in my life - I can diagnose the cause without even looking into the code. I'm not hating @inject or anything, it does the job, yet it caused me enough discomfort.

[TojikCZ]

Your API layer looks exactly like what we set out to do. Our old one is a mess and we dislike it.
Seems like my approach has crashed and burned. My component test is influencing the DI of an integration test.
So yea, going to do the injecting in the API layer, everything else is going to be pure python classes like you said

[ZipFile]

It's a bit tricky to cook DI for testing, but here's setup that works for me for non-unit tests: https://gist.github.com/ZipFile/7b9911c3f0c57c7d112725618a596464

[TojikCZ]

Thank you, I'll reference this when I get stuck doing these kinds of tests. So far, I have my service test without DI and moving forward. I burned an ungodly amount of time on this. Without you, I'd probably be forced to give up.

[TojikCZ]

Hi, so again, my nested structure has come back to bite me.
in main.py I save the RootContainer() to app.container, the recommended FastAPI way.
Then I inject using the relative path in an endpoint
printer_list_service: PrinterListService = Depends(Provide["printer_list_service"]
now in my test fixture, I try to get
app.container.main_package.printer_package.printer_repository()
I get two separate Singletons... That's usually not what happens with those 😅

These are integration tests using httpx, my limited understanding is it "simulates" a running web server while internally using just the FastAPI object

It seems like the injection which does not specify the full path attempts to instantiate a whole new PrinterContainer which depends on MainContainer and so it creates that too and the two identical containers coexist. I am unable to see the actual init calls as it all happens in some dynamic containers I cannot touch with my lowly human code.

More interestingly, when another endpoint needs the injection, the structure does not seem to be created a third time.

I printed them using this code:

    objs = gc.get_objects()
    for obj in objs:
        if isinstance(obj, DynamicContainer):
            print(f"Container {obj.parent_name}")

Container RootContainer
Container RootContainer.main_package
Container RootContainer.main_package.utils_package
Container RootContainer.main_package.printer_package
Container RootContainer.main_package
Container RootContainer.main_package.printer_package
Container RootContainer.main_package.utils_package
Container MainContainer.utils_package
Container MainContainer.printer_package

Nice, everything exists twice. The RootContainer seems to exist just once and some are rooted in MainContainer instead.

Doing it the full path way works, but the same duplicated DynamicContainer objects are printed
At least my code gets the same Singleton in the test and on @inject

Do you happen to know what do I keep unleashing upon myself? Like, what actually happens?
Sorry for bothering you again

[TojikCZ]

For now, I am opting for path aware containers. You give the class the dependency name and it gives you the full path, so re-structuring would mean one path change per container.

class ContainerWithPath(DeclarativeContainer):

    path: str

    @classmethod
    def dependency_path(cls, dependency_name: str) -> str:
        if cls.path is None:
            return dependency_name
        return f"{cls.path}.{dependency_name}"

but I am eager to get rid of this if anything else is, or becomes available

[ZipFile]

The recommended way aged poorly, unfortunately. It also makes mypy complain, and rightfully so. From my experience, the best way to avoid such kind of issues, is to avoid having any global instances of any "apps" (fastapi, flask, celery, etc...) at all and have them defined within the container, which on itself also should not have any global instances either.

Here's the missing part from the gist I shared before:

containers.py

Lifespan here is from dependency_injector.ext.starlette. It will call container.init_resources/container.shutdown_resources when your ASGI app starts and stops. This harmonizes well with _configure_loggging to setup logging, or any other kind of initialization stuff you need.

FeatureFlagsContainer is a bit synthetic example, we have in-house slightly-more complex implementation, but I've included it for demonstration purposes. In our code it is imported from internal library. It also has some Resources to init.

class FeatureFlagsContainer(DeclarativeContainer):
    api_key: Dependency[str] = Dependency(default="")
    enabled: Singleton[bool] = Singleton(bool, environment_key)
    cache_ttl: Dependency[float] = Dependency(default=120)
    cache: Singleton[FFCache] = Singleton(FFCache, cache_ttl=cache_ttl)
    client: Factory[FFClient] = Factory(FFClient, cache=cache, api_key=api_key)


class MainContainer(DeclarativeContainer):
    __self__ = Self()
    config: Configuration = Configuration()
    lifespan = Singleton(Lifespan, __self__)
    fastapi_app = Factory(
        make_fastapi_app,
        lifespan=lifespan ,
        title=config.misc.title,
        ...
    )
    _configure_loggging: Resource[None] = Resource(
        logging_init,
        log_config_file_path=config.logging.config_file_path,
        log_level=config.logging.level,
        log_config_name=config.logging.name,
    )
    ff = Container(
        FeatureFlagsContainer,  # Used as Provide["ff.client"] in code.
        api_key=config.ff.api_key,
        cache_ttl=config.ff.cache_ttl,
    )
    ...

application.py

Here exists everything FastAPI-related to avoid polluting other modules with API-layer stuff. It is separated purely for organizational reasons.

def make_fastapi_app(lifespan: Lifespan, title: str) -> FastAPI:
    app = FastAPI(
        title=title,
        lifespan=lifespan,
    )
    app.include_router(...)
    return app

asgi.py

This is a special file for uvicorn or any other ASGI server implementations. You never import it in your code for any reason but to launch the app.

def build_app() -> FastAPI:
    container = MainContainer()
    settings = Settings()
    container.config.from_dict(settings.model_dump())

    return container.fastapi_app()

__main__.py

This is actual web server launcher meant to be run as python -m myproj. We do not run our web server using uvicorn --factory myproject.asgi:build_app because when you run uvicorn as a CLI, it is impossible to convince it to not to do logging initialization, we're doing it manually in the container via Reasource provider.

import os

import uvicorn


def run() -> None:
    uvicorn.run(
        "myproject.asgi:build_app",
        factory=True,
        port=8000,
        host="0.0.0.0",
        reload=os.environ.get("ENVIRONMENT") == "local",
        log_config=None,  # this tells uvicorn to not to configure logging
    )


if __name__ == "__main__":
    run()

conftest.py

Self-explanatory. Tweak client fixture to your needs (e.g. include auth headers, other container provider override fixtures, etc...).

@fixture
def fastapi_app(
    container: Container[MainContainer],
    db_session: AsyncSession,
) -> FastAPI:
    return container.fastapi_app()  # type: ignore[no-any-return]


@fixture
def client(fastapi_app: FastAPI) -> AsyncClient:
    return AsyncClient(
        transport=ASGITransport(app=fastapi_app),
        base_url="http://test",
    )

I cannot answer what exactly is wrong with your setup without looking at the code, but hope my DI cooking recipe works for you.