Lifespan state not implemented in pytest?

[jrwagz]

I'm attempting to implement lifespan state into my application, as descried by @Kludex in his YouTube Video.

The problem I am running into, is that the lifespan state works just fine with I launch the app with uvicorn, or even gunicorn, however when I try and test the application with pytest, I'm getting the following error when I try and access request.state.cache:

AttributeError: 'State' object has no attribute 'cache'

Is anyone else seeing this?

Here are the versions that I was able to replicate this with:

$ python3                      
Python 3.11.9 (main, Apr  2 2024, 08:25:04) [Clang 15.0.0 (clang-1500.3.9.4)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import fastapi
>>> import starlette
>>> fastapi.__version__
'0.111.0'
>>> starlette.__version__
'0.37.2'

If I set a breakpoint, and inspect the objects, I see that the state object is indeed empty:

-> return request.state.cache
(Pdb) request
<starlette.requests.Request object at 0x103e2e690>
(Pdb) request.state
<starlette.datastructures.State object at 0x103e44990>
(Pdb) request.state.__dict__
{'_state': {}}

I even came up with a very simply test application to replicate the problem, based mainly on the starlette lifespan instructions.

Save the below as lifespan_state/main.py and then replicate the issue with:

pytest lifespan_state/main.py

Here is the example:

import contextlib
from typing import AsyncGenerator, AsyncIterator, Generator, TypedDict

from fastapi import FastAPI, Request
from fastapi.responses import PlainTextResponse
from fastapi.testclient import TestClient
import httpx
import pytest


class State(TypedDict):
    cache: httpx.AsyncClient


@contextlib.asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    async with httpx.AsyncClient() as client:
        yield {"cache": client}


app = FastAPI(
    lifespan=lifespan,
)


def client_from_request(request: Request) -> httpx.AsyncClient:
    return request.state.cache


@app.get("/")
async def homepage(request: Request) -> PlainTextResponse:
    client = client_from_request(request)
    response = await client.get("https://pastebin.com/")
    return PlainTextResponse(response.text)


@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"


@pytest.fixture()
def client() -> Generator:
    with TestClient(app) as client:
        yield client


@pytest.fixture()
async def async_client(client: TestClient) -> AsyncGenerator:
    async with httpx.AsyncClient(app=app, base_url=client.base_url) as ac:
        yield ac


@pytest.mark.anyio
async def test_homepage(async_client: httpx.AsyncClient):
    response = await async_client.get("/")
    assert response.status_code == 200

[Kludex]

https://github.com/kludex/fastapi-tips#5-use-httpxs-asyncclient-instead-of-testclient

[jrwagz]

Thank you!

I'll spell out the exact change to fix this, for the benefit of future internet travelers.

Specifically, to get this to work properly, you'll need to use the LifespanManager class from the asgi-lifespan package.

This means to add this import:

from asgi_lifespan import LifespanManager

and then modify the async_client to look like this:

@pytest.fixture()
async def async_client(client: TestClient) -> AsyncGenerator:
    async with LifespanManager(app) as manager:
        async with httpx.AsyncClient(app=manager.app, base_url=client.base_url) as ac:
            yield ac

Once you make those changes, running pytest lifespan_state/main.py should pass without error.

$ pytest lifespan_state/main.py
====================================================================== test session starts ======================================================================
platform darwin -- Python 3.11.9, pytest-8.2.0, pluggy-1.5.0 -- ~/test/.venv/bin/python
cachedir: .pytest_cache
rootdir: ~/test
configfile: pyproject.toml
plugins: anyio-4.3.0
collected 1 item                                                                                                                                                

lifespan_state/main.py::test_homepage PASSED                                                                                                              [100%]

======================================================================= 1 passed in 1.68s =======================================================================

Here is the full testcase again, with the above fixes applied:

import contextlib
from typing import AsyncGenerator, AsyncIterator, Generator, TypedDict

from asgi_lifespan import LifespanManager
from fastapi import FastAPI, Request
from fastapi.responses import PlainTextResponse
from fastapi.testclient import TestClient
import httpx
import pytest


class State(TypedDict):
    cache: httpx.AsyncClient


@contextlib.asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    async with httpx.AsyncClient() as client:
        yield {"cache": client}


app = FastAPI(
    lifespan=lifespan,
)


def client_from_request(request: Request) -> httpx.AsyncClient:
    return request.state.cache


@app.get("/")
async def homepage(request: Request) -> PlainTextResponse:
    client = client_from_request(request)
    response = await client.get("https://pastebin.com/")
    return PlainTextResponse(response.text)


@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"


@pytest.fixture()
def client() -> Generator:
    with TestClient(app) as client:
        yield client


@pytest.fixture()
async def async_client(client: TestClient) -> AsyncGenerator:
    async with LifespanManager(app) as manager:
        async with httpx.AsyncClient(app=manager.app, base_url=client.base_url) as ac:
            yield ac


@pytest.mark.anyio
async def test_homepage(async_client: httpx.AsyncClient):
    response = await async_client.get("/")
    assert response.status_code == 200