Can add state parameter in httpx.ASGITransport?

[Provinm]

recently i am making a TestClient based on httpx.AsyncClient, like what starlette have done

my raw code is

class Requestfactory(httpx.AsyncClient):

    def __init__(
        self,
        app: ASGIapp,
        app_state: t.Dict[str, t.Any] | None = None,
        lifespan_on: bool = True,
        base_url: str = "http://testserver",
    ) -> None:
        self.app = app
        if not app_state:
            app_state = {}
        app_state.update(self.app.state._state)
        transport = httpx.ASGITransport(app)
        scope: Scope = {
            "type": "lifespan",
            "asgi": {"version": "3.0", "spec_version": "2.1"},
            "state": app_state,
        }
        self.communicator = ApplicationCommunicator(app, scope)  # type: ignore
        self.lifespan_on = lifespan_on
        super().__init__(base_url=base_url, transport=transport)

the question is there is no way to set state in httpx.ASGITransport

so i have to re-write a ASGITransport class, but only add a state parameter.

class ASGITransport(httpx.AsyncBaseTransport):
    """
    Add state support to httpx.ASGITransport
    """

    def __init__(
        self,
        app: ASGIApp,
        raise_app_exceptions: bool = True,
        root_path: str = "",
        client: tuple[str, int] = ("127.0.0.1", 123),
        state: t.Dict[str, t.Any] | None = None,
    ) -> None:
        self.app = app
        self.raise_app_exceptions = raise_app_exceptions
        self.root_path = root_path
        self.client = client
        self.state = state or {}

    async def handle_async_request(
        self,
        request: Request,
    ) -> Response:
        assert isinstance(request.stream, httpx.AsyncByteStream)

        # ASGI scope.
        scope = {
            "type": "http",
            "asgi": {"version": "3.0"},
            "http_version": "1.1",
            "method": request.method,
            "headers": [(k.lower(), v) for (k, v) in request.headers.raw],
            "scheme": request.url.scheme,
            "path": request.url.path,
            "raw_path": request.url.raw_path.split(b"?")[0],
            "query_string": request.url.query,
            "server": (request.url.host, request.url.port),
            "client": self.client,
            "root_path": self.root_path,
            "state": self.state,
        }


        .......

so i was wondering if there's a chance to add state in httpx.ASGITransport

[Provinm]

refer: https://asgi.readthedocs.io/en/latest/specs/lifespan.html

[ketpatil77]

For state yielded through ASGI lifespan, you can avoid copying HTTPX's transport implementation: run the lifespan with asgi-lifespan and give ASGITransport manager.app, rather than the original app.

This minimal example shows startup state reaching an HTTP request:

import asyncio
import httpx
from asgi_lifespan import LifespanManager

async def app(scope, receive, send):
    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                scope["state"]["status"] = "ready"
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                await send({"type": "lifespan.shutdown.complete"})
                return
    elif scope["type"] == "http":
        body = scope["state"]["status"].encode()
        await send({"type": "http.response.start", "status": 200, "headers": []})
        await send({"type": "http.response.body", "body": body})

async def main():
    async with LifespanManager(app) as manager:
        transport = httpx.ASGITransport(app=manager.app)
        async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
            response = await client.get("/")
            assert response.text == "ready"

asyncio.run(main())

Install asgi-lifespan alongside HTTPX. The manager's wrapped app handles state propagation; using the original app in the transport skips that wrapper. HTTPX's ASGI transport does not itself run lifespan startup/shutdown.

Keep framework-wide app.state distinct from ASGI lifespan/request scope["state"]. The example addresses the latter; it does not introduce a new state= parameter on ASGITransport or reproduce every feature of Starlette's TestClient. For manually supplied per-request state unrelated to lifespan, a separate ASGI middleware wrapper can inject it without maintaining a fork of the transport.

References: asgi-lifespan state usage, HTTPX ASGI startup/shutdown.

AI-assisted answer using OpenAI Codex. Tested locally with Python 3.11.9, HTTPX 0.28.1, and asgi-lifespan 2.1.0; startup, HTTP state access, and shutdown complete successfully.