FastAPI routes wired to a dependency-injector DeclarativeContainer, tested offline: working versions (tested 2026-10-16)

fastapi==0.143.2
dependency-injector==4.49.1
httpx==0.28.1
pytest==9.1.1
pytest-asyncio==1.4.0

from __future__ import annotations

from dependency_injector import containers, providers


class TodoStore:
    def __init__(self) -> None:
        self._items: dict[int, str] = {}

    def add(self, title: str) -> int:
        todo_id = len(self._items) + 1
        self._items[todo_id] = title
        return todo_id

    def items(self) -> dict[int, str]:
        return dict(self._items)


class TodoService:
    def __init__(self, store: TodoStore, limit: int) -> None:
        self.store = store
        self.limit = limit

    def create(self, title: str) -> int:
        if len(self.store.items()) >= self.limit:
            raise ValueError("todo limit reached")
        return self.store.add(title)

    def listing(self) -> list[dict[str, int | str]]:
        return [{"id": i, "title": t} for i, t in sorted(self.store.items().items())]


class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    store = providers.Singleton(TodoStore)

    service = providers.Factory(
        TodoService,
        store=store,
        limit=config.limit.as_int(),
    )

from __future__ import annotations

from dependency_injector.wiring import Provide, inject
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel

from container import Container, TodoService

container = Container()
container.config.from_dict({"limit": 2})

app = FastAPI()


class TodoIn(BaseModel):
    title: str


class TodoOut(BaseModel):
    id: int


@app.post("/todos", response_model=TodoOut)
@inject
def create_todo(
    payload: TodoIn,
    service: TodoService = Depends(Provide[Container.service]),
) -> TodoOut:
    try:
        return TodoOut(id=service.create(payload.title))
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@app.get("/todos")
@inject
def list_todos(
    service: TodoService = Depends(Provide[Container.service]),
) -> list[dict[str, int | str]]:
    return service.listing()


container.wire(modules=[__name__])

from __future__ import annotations

import httpx
import pytest
from dependency_injector import providers

from app import app, container
from container import TodoService


@pytest.fixture(autouse=True)
def reset_container():
    container.config.reset_override()
    container.config.from_dict({"limit": 2})
    yield
    container.config.reset_override()
    container.config.from_dict({"limit": 2})
    container.store.reset_override()
    container.service.reset_override()
    container.reset_singletons()


def make_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test")


async def test_create_and_list_shares_singleton_store():
    async with make_client() as client:
        assert (await client.post("/todos", json={"title": "a"})).json() == {"id": 1}
        assert (await client.post("/todos", json={"title": "b"})).json() == {"id": 2}
        r = await client.get("/todos")
    assert r.json() == [{"id": 1, "title": "a"}, {"id": 2, "title": "b"}]


async def test_config_limit_returns_409():
    container.config.limit.override(1)
    async with make_client() as client:
        assert (await client.post("/todos", json={"title": "a"})).status_code == 200
        r = await client.post("/todos", json={"title": "b"})
    assert r.status_code == 409
    assert r.json()["detail"] == "todo limit reached"


async def test_override_store_propagates_to_service():
    class FakeStore:
        def items(self):
            return {7: "seeded"}

        def add(self, title):
            return 8

    container.store.override(providers.Singleton(FakeStore))
    async with make_client() as client:
        r = await client.get("/todos")
    assert r.json() == [{"id": 7, "title": "seeded"}]


async def test_override_service_then_reset():
    class FakeService:
        def create(self, title):
            return 42

        def listing(self):
            return [{"id": 42, "title": "mocked"}]

    container.service.override(providers.Factory(FakeService))
    async with make_client() as client:
        assert (await client.post("/todos", json={"title": "x"})).json() == {"id": 42}
    assert container.service.overridden
    container.service.reset_override()
    assert not container.service.overridden
    assert isinstance(container.service(), TodoService)


async def test_unwire_stops_injection():
    container.unwire()
    try:
        async with make_client() as client:
            with pytest.raises(AttributeError):
                await client.post("/todos", json={"title": "a"})
    finally:
        container.wire(modules=["app"])

[pytest]
pythonpath = .
asyncio_mode = auto