Merge branch 'master' into fix-content-length-validation

Author: jmestwa-coder

CHANGES/12358.misc.rst

@@ -0,0 +1 @@
+Changed ``zlib_executor_size`` default so compressed payloads are async by default -- by :user:`Dreamsorcerer`.

aiohttp/compression_utils.py

@@ -326,9 +326,15 @@ def decompress_sync(
     ) -> bytes:
         """Decompress the given data."""
         if hasattr(self._obj, "decompress"):
-            result = cast(bytes, self._obj.decompress(data, max_length))
+            if max_length == ZLIB_MAX_LENGTH_UNLIMITED:
+                result = cast(bytes, self._obj.decompress(data))
+            else:
+                result = cast(bytes, self._obj.decompress(data, max_length))
         else:
-            result = cast(bytes, self._obj.process(data, max_length))
+            if max_length == ZLIB_MAX_LENGTH_UNLIMITED:
+                result = cast(bytes, self._obj.process(data))
+            else:
+                result = cast(bytes, self._obj.process(data, max_length))
         # Only way to know that brotli has no further data is checking we get no output
         self._last_empty = result == b""
         return result

aiohttp/http_parser.py

@@ -2,6 +2,7 @@
 import asyncio
 import re
 import string
+import sys
 from contextlib import suppress
 from enum import IntEnum
 from re import Pattern
@@ -1118,10 +1119,12 @@ def feed_data(self, chunk: bytes) -> bool:
                 encoding=self.encoding, suppress_deflate_header=True
             )
 
+        low_water = self.out._low_water
+        max_length = (
+            0 if low_water >= sys.maxsize else max(self._max_decompress_size, low_water)
+        )
         try:
-            chunk = self.decompressor.decompress_sync(
-                chunk, max_length=self._max_decompress_size
-            )
+            chunk = self.decompressor.decompress_sync(chunk, max_length=max_length)
         except Exception:
             raise ContentEncodingError(
                 "Can not decode content-encoding: %s" % self.encoding

aiohttp/streams.py

@@ -1,5 +1,6 @@
 import asyncio
 import collections
+import sys
 import warnings
 from collections.abc import Awaitable, Callable
 from typing import Final, Generic, TypeVar
@@ -67,31 +68,7 @@ async def __anext__(self) -> tuple[bytes, bool]:
         return rv
 
 
-class AsyncStreamReaderMixin:
-
-    __slots__ = ()
-
-    def __aiter__(self) -> AsyncStreamIterator[bytes]:
-        return AsyncStreamIterator(self.readline)  # type: ignore[attr-defined]
-
-    def iter_chunked(self, n: int) -> AsyncStreamIterator[bytes]:
-        """Returns an asynchronous iterator that yields chunks of size n."""
-        return AsyncStreamIterator(lambda: self.read(n))  # type: ignore[attr-defined]
-
-    def iter_any(self) -> AsyncStreamIterator[bytes]:
-        """Yield all available data as soon as it is received."""
-        return AsyncStreamIterator(self.readany)  # type: ignore[attr-defined]
-
-    def iter_chunks(self) -> ChunkTupleAsyncStreamIterator:
-        """Yield chunks of data as they are received by the server.
-
-        The yielded objects are tuples
-        of (bytes, bool) as returned by the StreamReader.readchunk method.
-        """
-        return ChunkTupleAsyncStreamIterator(self)  # type: ignore[arg-type]
-
-
-class StreamReader(AsyncStreamReaderMixin):
+class StreamReader:
     """An enhancement of asyncio.StreamReader.
 
     Supports asynchronous iteration by line, chunk or as available::
@@ -174,9 +151,35 @@ def __repr__(self) -> str:
             info.append("e=%r" % self._exception)
         return "<%s>" % " ".join(info)
 
+    def __aiter__(self) -> AsyncStreamIterator[bytes]:
+        return AsyncStreamIterator(self.readline)
+
+    def iter_chunked(self, n: int) -> AsyncStreamIterator[bytes]:
+        """Returns an asynchronous iterator that yields chunks of size n."""
+        self.set_read_chunk_size(n)
+        return AsyncStreamIterator(lambda: self.read(n))
+
+    def iter_any(self) -> AsyncStreamIterator[bytes]:
+        """Yield all available data as soon as it is received."""
+        return AsyncStreamIterator(self.readany)
+
+    def iter_chunks(self) -> ChunkTupleAsyncStreamIterator:
+        """Yield chunks of data as they are received by the server.
+
+        The yielded objects are tuples
+        of (bytes, bool) as returned by the StreamReader.readchunk method.
+        """
+        return ChunkTupleAsyncStreamIterator(self)
+
     def get_read_buffer_limits(self) -> tuple[int, int]:
         return (self._low_water, self._high_water)
 
+    def set_read_chunk_size(self, n: int) -> None:
+        """Raise buffer limits to match the consumer's chunk size."""
+        if n > self._low_water:
+            self._low_water = n
+            self._high_water = n * 2
+
     def exception(self) -> type[BaseException] | BaseException | None:
         return self._exception
 
@@ -410,10 +413,8 @@ async def read(self, n: int = -1) -> bytes:
             return b""
 
         if n < 0:
-            # This used to just loop creating a new waiter hoping to
-            # collect everything in self._buffer, but that would
-            # deadlock if the subprocess sends more than self.limit
-            # bytes.  So just call self.readany() until EOF.
+            # Reading everything — remove decompression chunk limit.
+            self.set_read_chunk_size(sys.maxsize)
             blocks = []
             while True:
                 block = await self.readany()
@@ -422,6 +423,7 @@ async def read(self, n: int = -1) -> bytes:
                 blocks.append(block)
             return b"".join(blocks)
 
+        self.set_read_chunk_size(n)
         # TODO: should be `if` instead of `while`
         # because waiter maybe triggered on chunk end,
         # without feeding any data
@@ -595,6 +597,9 @@ async def wait_eof(self) -> None:
     def feed_data(self, data: bytes) -> bool:
         return False
 
+    def set_read_chunk_size(self, n: int) -> None:
+        return
+
     async def readline(self, *, max_line_length: int | None = None) -> bytes:
         return b""
 

aiohttp/web_request.py

@@ -628,6 +628,10 @@ async def read(self) -> bytes:
         Returns bytes object with full request content.
         """
         if self._read_bytes is None:
+            # Raise the buffer limits so compressed payloads decompress in
+            # larger chunks instead of many small pause/resume cycles.
+            if self._client_max_size:
+                self._payload.set_read_chunk_size(self._client_max_size)
             body = bytearray()
             while True:
                 chunk = await self._payload.readany()

aiohttp/web_response.py

@@ -14,7 +14,7 @@
 
 from . import hdrs, payload
 from .abc import AbstractStreamWriter
-from .compression_utils import ZLibCompressor
+from .compression_utils import MAX_SYNC_CHUNK_SIZE, ZLibCompressor
 from .helpers import (
     ETAG_ANY,
     QUOTED_ETAG_RE,
@@ -35,7 +35,6 @@
 from .typedefs import JSONBytesEncoder, JSONEncoder, LooseHeaders
 
 REASON_PHRASES = {http_status.value: http_status.phrase for http_status in HTTPStatus}
-LARGE_BODY_SIZE = 1024**2
 
 __all__ = (
     "ContentCoding",
@@ -547,7 +546,7 @@ def __init__(
         headers: LooseHeaders | None = None,
         content_type: str | None = None,
         charset: str | None = None,
-        zlib_executor_size: int | None = None,
+        zlib_executor_size: int = MAX_SYNC_CHUNK_SIZE,
         zlib_executor: Executor | None = None,
     ) -> None:
         if body is not None and text is not None:
@@ -726,13 +725,6 @@ async def _do_start_compression(self, coding: ContentCoding) -> None:
             executor=self._zlib_executor,
         )
         assert self._body is not None
-        if self._zlib_executor_size is None and len(self._body) > LARGE_BODY_SIZE:
-            warnings.warn(
-                "Synchronous compression of large response bodies "
-                f"({len(self._body)} bytes) might block the async event loop. "
-                "Consider providing a custom value to zlib_executor_size/"
-                "zlib_executor response properties or disabling compression on it."
-            )
         self._compressed_body = (
             await compressor.compress(self._body) + compressor.flush()
         )

docs/testing.rst

@@ -537,6 +537,12 @@ basis, the TestClient object can be used directly::
 A full list of the utilities provided can be found at the
 :data:`api reference <aiohttp.test_utils>`
 
+For end-to-end client code that talks to an external service, it is
+recommended to run a small fake server rather than patching private aiohttp
+internals. The ``examples/fake_server.py`` demo shows such an approach: start
+a local :class:`~aiohttp.web.Application`, point a custom resolver at it, and
+exercise the client against that controlled endpoint.
+
 
 Testing API Reference
 ---------------------

requirements/base-ft.txt

@@ -30,7 +30,7 @@ multidict==6.7.1
     # via
     #   -r requirements/runtime-deps.in
     #   yarl
-packaging==26.0
+packaging==26.1
     # via gunicorn
 propcache==0.4.1
     # via

requirements/base.txt

@@ -30,7 +30,7 @@ multidict==6.7.1
     # via
     #   -r requirements/runtime-deps.in
     #   yarl
-packaging==26.0
+packaging==26.1
     # via gunicorn
 propcache==0.4.1
     # via

requirements/constraints.txt

@@ -34,7 +34,7 @@ blockbuster==1.5.26
     #   -r requirements/test-common.in
 brotli==1.2.0 ; platform_python_implementation == "CPython"
     # via -r requirements/runtime-deps.in
-build==1.4.2
+build==1.4.3
     # via pip-tools
 certifi==2026.2.25
     # via requests
@@ -69,7 +69,7 @@ exceptiongroup==1.3.1
     # via pytest
 execnet==2.1.2
     # via pytest-xdist
-filelock==3.25.2
+filelock==3.28.0
     # via
     #   python-discovery
     #   virtualenv
@@ -125,7 +125,7 @@ mypy-extensions==1.1.0
     # via mypy
 nodeenv==1.10.0
     # via pre-commit
-packaging==26.0
+packaging==26.1
     # via
     #   build
     #   gunicorn
@@ -160,9 +160,9 @@ pycares==5.0.1
     # via aiodns
 pycparser==3.0
     # via cffi
-pydantic==2.13.0
+pydantic==2.13.2
     # via python-on-whales
-pydantic-core==2.46.0
+pydantic-core==2.46.2
     # via pydantic
 pyenchant==3.3.0
     # via sphinxcontrib-spelling
@@ -183,7 +183,7 @@ pytest==9.0.3
     #   pytest-cov
     #   pytest-mock
     #   pytest-xdist
-pytest-codspeed==4.3.0
+pytest-codspeed==4.4.0
     # via
     #   -r requirements/lint.in
     #   -r requirements/test-common.in
@@ -281,7 +281,7 @@ uvloop==0.21.0 ; platform_system != "Windows"
     #   -r requirements/lint.in
 valkey==6.1.1
     # via -r requirements/lint.in
-virtualenv==21.2.0
+virtualenv==21.2.4
     # via pre-commit
 wait-for-it==2.3.0
     # via -r requirements/test-common.in