GeezerVison

status. `Content-Length` may reject early but never admit early, because it reports the compressed

size and a 1000:1 gzip bomb passes a header check by construction. It replaced an error-only

pre-check, since memory exhaustion has no status code, and it never applies to caller-driven

`stream()` iteration, whose purpose is bodies too large to buffer. The response is rebuilt

through `httpx2.Response(content=...)` with public API only: `elapsed`, `next_request` and

`default_encoding` are copied over, and `num_bytes_downloaded`, which has no public setter, reads 0.

Counting bytes in a wrapper around the response stream would avoid the rebuild; that is deferred to

[#170](https://github.com/modern-python/httpware/issues/170).

`stream()` iteration, whose purpose is bodies too large to buffer. For a body with no

`Content-Encoding` the decoded bytes are the wire bytes, so the cap wraps the public

`Response.stream` in a stream that counts bytes and raises once the count passes the cap, and httpx2

reads the response with its own `read()`. The response stays the one httpx2 built. An encoded body

is read through `iter_bytes()` and rebuilt with `httpx2.Response(content=...)`, using public API

only: `elapsed`, `next_request` and `default_encoding` are copied over, and `num_bytes_downloaded`,

which has no public setter, reads 0.

Both clients accept `max_response_body_bytes: int | None = None`. By default there is no limit. When it is set, a response body larger than the cap raises `ResponseTooLargeError` instead of being returned, whatever the status: a `200` trips it as easily as a `500`. The cap counts decoded bytes, after decompression. It applies to `send()` and the verb methods, and to the error body that `stream()` reads before raising a `StatusError`. Bytes you read yourself while iterating a `stream()` are never capped, except with an `auth` that sets `requires_response_body`: that auth needs every body, so `stream()` reads the final one under the cap before yielding it. With a cap set, httpware runs the client's `auth` flow and, with `follow_redirects=True`, follows the redirects itself, and caps only the final response. It reads the body of each intermediate response, such as a redirect or a `DigestAuth` challenge, and throws it away, so the connection can be reused and the responses in `response.history` have no content. An intermediate body declared or read past the cap is not an error: httpware stops reading it and closes that connection. An `auth` that sets `requires_response_body`, like a token refresh that parses the token endpoint's JSON, gets each response read under the cap instead. Client `auth` is sent to the first URL only. `httpx2` keeps its `Authorization` header on a redirect within the same origin or from `http` to `https` on the same host, and drops it otherwise.

A response read under the cap is rebuilt from its decoded body. It keeps `elapsed`, `next_request` and the client's `default_encoding`. It loses the `Content-Encoding` and `Transfer-Encoding` headers, its `Content-Length` is the decoded size, and `num_bytes_downloaded` is `0`.

A response read under the cap keeps `elapsed`, `next_request` and the client's `default_encoding`. A response without `Content-Encoding` is the one `httpx2` built, with the same headers and `num_bytes_downloaded` it has without a cap. A response with `Content-Encoding` is rebuilt from its decoded body: it loses the `Content-Encoding` and `Transfer-Encoding` headers, its `Content-Length` is the decoded size, and `num_bytes_downloaded` is `0`.

`ResponseTooLargeError` carries:

import contextlib

import datetime as dt

import typing

from collections.abc import Mapping

from collections.abc import AsyncGenerator, AsyncIterator, Iterator, Mapping

from http import HTTPStatus

import httpx2

return content_length

class _CappedSyncStream(httpx2.SyncByteStream):

"""Wrap a response stream, raising ResponseTooLargeError(reason="streamed") once it yields more than `cap` bytes."""

def __init__(self, stream: httpx2.SyncByteStream, cap: int, status_code: int, content_length: int | None) -> None:

self._stream = stream

self._cap = cap

self._status_code = status_code

self._content_length = content_length

@property

def elapsed(self) -> dt.timedelta | None:

"""The wrapped stream's `elapsed`, which httpx2 reads back as `Response.elapsed`."""

return getattr(self._stream, "elapsed", None)

def __iter__(self) -> Iterator[bytes]:

read = 0

for chunk in self._stream:

read += len(chunk)

if read > self._cap:

raise ResponseTooLargeError(

status_code=self._status_code,

limit=self._cap,

content_length=self._content_length,

reason="streamed",

)

yield chunk

def close(self) -> None:

self._stream.close()

class _CappedAsyncStream(httpx2.AsyncByteStream):

"""Async mirror of `_CappedSyncStream`."""

def __init__(self, stream: httpx2.AsyncByteStream, cap: int, status_code: int, content_length: int | None) -> None:

self._stream = stream

self._cap = cap

self._status_code = status_code

self._content_length = content_length

@property

def elapsed(self) -> dt.timedelta | None:

"""The wrapped stream's `elapsed`, which httpx2 reads back as `Response.elapsed`."""

return getattr(self._stream, "elapsed", None)

async def __aiter__(self) -> AsyncIterator[bytes]:

chunks = self._stream.__aiter__()

try:

read = 0

async for chunk in chunks:

read += len(chunk)

if read > self._cap:

raise ResponseTooLargeError(

status_code=self._status_code,

limit=self._cap,

content_length=self._content_length,

reason="streamed",

)

yield chunk

finally:

if isinstance(chunks, AsyncGenerator):

await chunks.aclose()

async def aclose(self) -> None:

await self._stream.aclose()

def _build_buffered_response(response: httpx2.Response, content: bytes, request: httpx2.Request) -> httpx2.Response:

"""Rebuild a buffered Response from the closed streaming `response` and its final decoded `content`."""

buffered = httpx2.Response(

if not _response_has_body(request.method, response.status_code):

response.read() # empty body; preserve the original response (and its headers)

return response

content_length = _reject_if_declared_length_exceeds_cap(response, cap)

if "content-encoding" not in response.headers:

stream = typing.cast("httpx2.SyncByteStream", response.stream)

response.stream = _CappedSyncStream(stream, cap, response.status_code, content_length)

try:

response.read()

except BaseException:

response.close()

raise

return response

try:

content = _accumulate_capped(response.iter_bytes(), cap)

except _CapExceeded:

await response.aread() # empty body; preserve the original response (and its headers)

return response

content_length = _reject_if_declared_length_exceeds_cap(response, cap)

if "content-encoding" not in response.headers:

stream = typing.cast("httpx2.AsyncByteStream", response.stream)

response.stream = _CappedAsyncStream(stream, cap, response.status_code, content_length)

try:

await response.aread()

except BaseException:

await response.aclose()

raise

return response

buf = bytearray()

async for chunk in response.aiter_bytes():

buf += chunk

import gzip

from collections.abc import AsyncIterator

from collections.abc import AsyncIterator, Iterator

import httpx2

import pytest

client.close()

def test_read_capped_returns_the_same_response_for_an_unencoded_body() -> None:

body = b"hello world"

def handler(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001

return httpx2.Response(200, content=iter([body]))

client, resp = _sync_stream(handler)

try:

out = _read_capped(resp, 1000, resp.request)

assert out is resp

assert out.content == body

assert out.is_closed

assert out.num_bytes_downloaded == len(body)

assert out.headers["transfer-encoding"] == "chunked"

finally:

resp.close()

client.close()

def test_read_capped_rebuilds_the_response_for_an_encoded_body() -> None:

raw = gzip.compress(b"A" * 500)

def handler(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001

return httpx2.Response(200, headers={"content-encoding": "gzip"}, content=raw)

client, resp = _sync_stream(handler)

try:

out = _read_capped(resp, 1_000_000, resp.request)

assert out is not resp

assert out.num_bytes_downloaded == 0

finally:

resp.close()

client.close()

def test_read_capped_unencoded_over_cap_closes_the_stream() -> None:

closed: list[bool] = []

class _Body(httpx2.SyncByteStream):

def __iter__(self) -> Iterator[bytes]:

yield b"a" * 50

yield b"b" * 50

def close(self) -> None:

closed.append(True)

def handler(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001

return httpx2.Response(200, stream=_Body())

client, resp = _sync_stream(handler)

try:

with pytest.raises(ResponseTooLargeError) as caught:

_read_capped(resp, 70, resp.request)

assert caught.value.reason == "streamed"

assert caught.value.limit == 70 # noqa: PLR2004 — cap above

assert caught.value.status_code == 200 # noqa: PLR2004 — mirrors handler

assert resp.is_closed

assert closed == [True]

finally:

resp.close()

client.close()

await client.aclose()

async def test_read_capped_async_returns_the_same_response_for_an_unencoded_body() -> None:

body = b"hello world"

async def chunks() -> AsyncIterator[bytes]:

yield body

def handler(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001

return httpx2.Response(200, content=chunks())

client, resp = await _async_stream(handler)

try:

out = await _read_capped_async(resp, 1000, resp.request)

assert out is resp

assert out.content == body

assert out.is_closed

assert out.num_bytes_downloaded == len(body)

finally:

await resp.aclose()

await client.aclose()

async def test_read_capped_async_rebuilds_the_response_for_an_encoded_body() -> None:

raw = gzip.compress(b"A" * 500)

def handler(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001

return httpx2.Response(200, headers={"content-encoding": "gzip"}, content=raw)

client, resp = await _async_stream(handler)

try:

out = await _read_capped_async(resp, 1_000_000, resp.request)

assert out is not resp

assert out.num_bytes_downloaded == 0

finally:

await resp.aclose()

await client.aclose()

async def test_read_capped_async_unencoded_over_cap_closes_the_stream() -> None:

closed: list[bool] = []

class _Body(httpx2.AsyncByteStream):

async def __aiter__(self) -> AsyncIterator[bytes]:

yield b"a" * 50

yield b"b" * 50

async def aclose(self) -> None:

closed.append(True)

def handler(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001

return httpx2.Response(200, stream=_Body())

client, resp = await _async_stream(handler)

try:

with pytest.raises(ResponseTooLargeError) as caught:

await _read_capped_async(resp, 70, resp.request)

assert caught.value.reason == "streamed"

assert caught.value.limit == 70 # noqa: PLR2004 — cap above

assert resp.is_closed

assert closed == [True]

finally:

await resp.aclose()

await client.aclose()

) as client:

response = await client.get("https://example.test/x")

assert response.text == "café"

def test_sync_response_keeps_num_bytes_downloaded_with_or_without_a_cap(cap: int | None) -> None:

with Client(transport=httpx2.MockTransport(_streamed_body), max_response_body_bytes=cap) as client:

response = client.get("https://example.test/x")

assert response.num_bytes_downloaded == len(b"hello")

async def test_async_response_keeps_num_bytes_downloaded_with_or_without_a_cap(cap: int | None) -> None:

async with AsyncClient(transport=httpx2.MockTransport(_streamed_body_async), max_response_body_bytes=cap) as client:

response = await client.get("https://example.test/x")

assert response.num_bytes_downloaded == len(b"hello")