`AiohttpResponseStream` does not work if the `aiohttp` response was `read`

[jamesbraza]

I am trying to use httpx-aiohttp, and am hitting an incompatibility with vcrpy + httpx_aiohttp.AiohttpResponseStream. What happens is:

1. The inner aiohttp request has its response modified via this vcrpy patch to be read here.
   • vcrpy does this so the response can be stored in a VCR cassette.
2. That read (1) exhausts the aiohttp.StreamReader and (2) moves the data to aiohttp.ClientResponse._body (link).
3. The httpx.Response then returned from httpx_aiohttp.AiohttpTransport.handle_async_request is broken, because calls to aread return b"". Any downstream users of response.json() will get blown up due to json.loads("") throwing a JSONDecodeError.
   • This b"" return happens because the AiohttpResponseStream.__aiter__ doesn't look at _body (link).

Here is a minimal reproducer:

import httpx_aiohttp
import pytest

@pytest.mark.vcr
@pytest.mark.asyncio
async def test_compressed_get() -> None:
    """Make a GET request, with gzip compression."""
    async with httpx_aiohttp.HttpxAiohttpClient(timeout=60) as client:
        # Can also use https://httpbin.org/gzip, but it can be flaky
        response = await client.get("https://httpbingo.org/gzip")
        response.raise_for_status()
        assert response.status_code == 200
        assert response.json()["gzipped"]

Running this code with Python 3.13, pytest==8.4.1, pytest-asyncio==1.1.0, pytest-recording==0.13.4, vcrpy==7.0.0, httpx==0.28.1, and httpx-aiohttp==0.1.8, httpcore==1.0.9, and no-prerecorded VCR cassette, the .json() call at the end leads to:

json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

If you realize, this is actually the same issue we recently solved with hishel: karpetrosyan/hishel#374

A possible starter solution for AiohttpResponseStream is:

class AiohttpResponseStream(httpx.AsyncByteStream):
    CHUNK_SIZE = 1024 * 16

    def __init__(self, aiohttp_response: ClientResponse) -> None:
        self._aiohttp_response = aiohttp_response

    async def __aiter__(self) -> typing.AsyncIterator[bytes]:
        with map_aiohttp_exceptions():
            if self._aiohttp_response._body is not None:
                # Happens if some intermediary called `await _aiohttp_response.read()`
                # TODO: take into account chunk size
                # TODO: take into account `not self._aiohttp_response.content.at_eof()`?
                yield self._aiohttp_response._body
            else:
                async for chunk in self._aiohttp_response.content.iter_chunked(self.CHUNK_SIZE):
                    yield chunk

And lastly, I would like to say thank you for making this super useful library! Happy it exists and looking forward to using it.

[karpetrosyan]

The httpx.Response then returned from httpx_aiohttp.AiohttpTransport.handle_async_request is broken, because calls to aread return b"". Any downstream users of response.json() will get blown up due to json.loads("") throwing a JSONDecodeError.

Yeah, that's bad. We need to try to read the body and if we fail, then fallback to stream, as we do with httpx request here