RuntimeError: The server does not support "state" in the lifespan scope.

[matt3o]

Heyho! I tried to get FastAPI streaming to work and according to multiple bug entries this tool here should be the way. However, when I tried it out, I got the errors below. I use the lifespan to inject services into the Starlette request state. Sadly this does appear not to work or I have missed an important step. I will paste the error log below. Thanks already for any help! Matthias

request = <SubRequest 'aclient' for <Coroutine test_streaming_chat_minimal>>, kwargs = {}, unittest = False, func = <function aclient at 0x7efdfb8f39c0>                                                                   15:58:40 [45/1931]
setup = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.setup at 0x7efdf13a4220>, finalizer = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.finalizer at 0x7efdf13a4180>                                                                                                                                                                                                                                                          @functools.wraps(fixture)                                                                                                                                                                                                                    def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):                                                                                                                                                                           unittest = fixturedef.unittest if hasattr(fixturedef, "unittest") else False                                                                                                                                                                 func = _perhaps_rebind_fixture_func(fixture, request.instance, unittest)                                                                                                                                                                     event_loop = kwargs.pop(event_loop_fixture_id)                                                                                                                                                                                               gen_obj = func(                                                                                                                                                                                                                                  **_add_kwargs(func, kwargs, event_loop_fixture_id, event_loop, request)                                                                                                                                                                  )
                                                                                                                                                                                                                                                     async def setup():
            res = await gen_obj.__anext__()
            return res

        def finalizer() -> None:
            """Yield again, to finalize."""

            async def async_finalizer() -> None:
                try:
                    await gen_obj.__anext__()
                except StopAsyncIteration:
                    pass
                else:
                    msg = "Async generator fixture didn't stop."
                    msg += "Yield only once."
                    raise ValueError(msg)

            event_loop.run_until_complete(async_finalizer())

>       result = event_loop.run_until_complete(setup())

../../.lvenv/lib/python3.12/site-packages/pytest_asyncio/plugin.py:347:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _/usr/lib/python3.12/asyncio/base_events.py:687: in run_until_complete
    return future.result()
../../.lvenv/lib/python3.12/site-packages/pytest_asyncio/plugin.py:329: in setup
    res = await gen_obj.__anext__()
tests/unit_tests/conftest.py:19: in aclient
    async with TestClient(app) as testclient:
../../.lvenv/lib/python3.12/site-packages/async_asgi_testclient/testing.py:91: in __aenter__                                                                                                                                                     await self.send_lifespan("startup")
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <async_asgi_testclient.testing.TestClient object at 0x7efdf13b14f0>, action = 'startup'

    async def send_lifespan(self, action):
        await self._lifespan_input_queue.put({"type": f"lifespan.{action}"})
        message = await receive(self._lifespan_output_queue, timeout=self.timeout)

        if isinstance(message, Message):
            raise Exception(f"{message.event} - {message.reason} - {message.task}")

        if message["type"] == f"lifespan.{action}.complete":
            pass
        elif message["type"] == f"lifespan.{action}.failed":
>           raise Exception(message)
E           Exception: {'type': 'lifespan.startup.failed', 'message': 'Traceback (most recent call last):\n  File "/mnt/c/Users//code//.lvenv/lib/python3.12/site-packages/starlette/routing.py", line 735, in lifespan\n    raise RuntimeError(\nRuntimeError: The server does not support "state" in the lifespan scope.\n'}

../../.lvenv/lib/python3.12/site-packages/async_asgi_testclient/testing.py:108: Exception

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    logger.info("Initializing App")
    state = initialize_services()
    async with state["http_client"]:
        logger.info("App state initialized")
        yield state
    logger.info("Shutting down App")

[matt3o]

Correction, I did not follow the instructions correctly. However, maybe this helps someone else too :D

You need a LifespanManager and you also need to pass that new object into httpx, then it does work as expected, see code below.

@pytest.fixture(scope="session")
async def aclient():
    async with LifespanManager(app) as manager:
        async with TestClient(manager.app) as testclient:
            yield testclient