Define explicit StreamingBody API, add httpx read(amt) support

CHANGES.rst

@@ -1,6 +1,19 @@
 Changes
 -------
 
+3.5.0 (2026-04-20)
+^^^^^^^^^^^^^^^^^^
+* define explicit ``StreamingBody`` API and add ``read(amt)`` support for the
+  httpx backend. ``StreamingBody`` and ``HttpxStreamingBody`` are now
+  ``AioStreamingBody`` (subclassing ``botocore.response.StreamingBody``) and
+  ``AioHttpxStreamingBody`` (subclassing ``AioStreamingBody``); the old names
+  remain as module-level aliases.
+* ``AioStreamingBody.__aenter__`` now returns ``self`` (previously returned the
+  raw aiohttp ``ClientResponse``). Use ``body.raw_stream`` for direct access.
+* ``HttpxStreamingChecksumBody.readinto`` no longer calls ``content.read`` on
+  an httpx ``Response`` (which would fail); it now delegates to the shared
+  ``_ChecksumMixin`` implementation.
+
 3.4.0 (2026-04-07)
 ^^^^^^^^^^^^^^^^^^
 * bump botocore dependency specification to support ``"botocore >= 1.42.79, < 1.42.85"``

CLAUDE.md

@@ -1,5 +1,7 @@
 # Environment
 
+Default branch is `main` (not `master`). Use `--base main` for PRs.
+
 Dependencies are managed by `uv`. Never use `pip install`.
 To read botocore source, find the installed path with
 `python3 -c "import botocore; print(botocore.__file__)"`,

aiobotocore/__init__.py

@@ -1 +1 @@
-__version__ = '3.4.0'
+__version__ = '3.5.0'

aiobotocore/endpoint.py

@@ -17,7 +17,7 @@
 from aiobotocore.httpchecksum import handle_checksum_body
 from aiobotocore.httpsession import AIOHTTPSession
 from aiobotocore.parsers import AioResponseParserFactory
-from aiobotocore.response import HttpxStreamingBody, StreamingBody
+from aiobotocore.response import AioHttpxStreamingBody, AioStreamingBody
 
 try:
     import httpx
@@ -54,11 +54,13 @@ async def convert_to_response_dict(http_response, operation_model):
     elif operation_model.has_event_stream_output:
         response_dict['body'] = http_response.raw
     elif operation_model.has_streaming_output:
+        length = response_dict['headers'].get('content-length')
         if httpx and isinstance(http_response.raw, httpx.Response):
-            response_dict['body'] = HttpxStreamingBody(http_response.raw)
+            response_dict['body'] = AioHttpxStreamingBody(
+                http_response.raw, length
+            )
         else:
-            length = response_dict['headers'].get('content-length')
-            response_dict['body'] = StreamingBody(http_response.raw, length)
+            response_dict['body'] = AioStreamingBody(http_response.raw, length)
     else:
         response_dict['body'] = await http_response.content
     return response_dict

aiobotocore/httpchecksum.py

@@ -12,7 +12,7 @@
 )
 
 from aiobotocore._helpers import resolve_awaitable
-from aiobotocore.response import HttpxStreamingBody, StreamingBody
+from aiobotocore.response import AioHttpxStreamingBody, AioStreamingBody
 
 try:
     import httpx
@@ -74,11 +74,13 @@ async def __anext__(self):
         raise StopAsyncIteration()
 
 
-# unfortunately we can't inherit from botocore's StreamingChecksumBody due to
-# subclassing
-class StreamingChecksumBody(StreamingBody):
-    def __init__(self, raw_stream, content_length, checksum, expected):
-        super().__init__(raw_stream, content_length)
+class _ChecksumMixin:
+    """Mixin that adds checksum validation to a StreamingBody.
+
+    Shared by both aiohttp and httpx checksum body classes.
+    """
+
+    def _init_checksum(self, checksum, expected):
         self._checksum = checksum
         self._expected = expected
 
@@ -90,9 +92,7 @@ async def read(self, amt=None):
         return chunk
 
     async def readinto(self, b: bytearray):
-        chunk = await self.__wrapped__.content.read(len(b))
-        amount_read = len(chunk)
-        b[:amount_read] = chunk
+        amount_read = await super().readinto(b)
 
         if amount_read == len(b):
             view = b
@@ -113,48 +113,40 @@ def _validate_checksum(self):
             raise FlexibleChecksumError(error_msg=error_msg)
 
 
-# TODO: fix inheritance? read & _validate_checksum are the exact same as above
-# only diff is super class and how to call __init__
-class HttpxStreamingChecksumBody(HttpxStreamingBody):
+class AioStreamingChecksumBody(_ChecksumMixin, AioStreamingBody):
+    """AioStreamingBody with checksum validation (aiohttp backend)."""
+
     def __init__(self, raw_stream, content_length, checksum, expected):
-        # HttpxStreamingbody doesn't use content_length
-        super().__init__(raw_stream)
-        self._checksum = checksum
-        self._expected = expected
+        super().__init__(raw_stream, content_length)
+        self._init_checksum(checksum, expected)
 
-    # TODO: this class is largely (or possibly entirely) untested. The tests need to be
-    # more thoroughly rewritten wherever they directly create Streamingbody,
-    # StreamingChecksumBody, etc.
 
-    async def read(self, amt=None):
-        chunk = await super().read(amt=amt)
-        self._checksum.update(chunk)
-        if amt is None or (not chunk and amt > 0):
-            self._validate_checksum()
-        return chunk
+class AioHttpxStreamingChecksumBody(_ChecksumMixin, AioHttpxStreamingBody):
+    """AioHttpxStreamingBody with checksum validation (httpx backend)."""
 
-    async def readinto(self, b: bytearray):
-        chunk = await self.__wrapped__.content.read(len(b))
-        amount_read = len(chunk)
-        b[:amount_read] = chunk
+    def __init__(self, raw_stream, content_length, checksum, expected):
+        super().__init__(raw_stream, content_length)
+        self._init_checksum(checksum, expected)
 
-        if amount_read == len(b):
-            view = b
-        else:
-            view = memoryview(b)[:amount_read]
 
-        self._checksum.update(view)
-        if amount_read == 0 and len(b) > 0:
-            self._validate_checksum()
-        return amount_read
+# Backwards-compatibility aliases for pre-Aio-prefix names.
+StreamingChecksumBody = AioStreamingChecksumBody
+HttpxStreamingChecksumBody = AioHttpxStreamingChecksumBody
 
-    def _validate_checksum(self):
-        if self._checksum.digest() != base64.b64decode(self._expected):
-            error_msg = (
-                f"Expected checksum {self._expected} did not match calculated "
-                f"checksum: {self._checksum.b64digest()}"
-            )
-            raise FlexibleChecksumError(error_msg=error_msg)
+
+def _handle_streaming_response(http_response, response, algorithm):
+    checksum_cls = _CHECKSUM_CLS.get(algorithm)
+    header_name = f"x-amz-checksum-{algorithm}"
+    if httpx is not None and isinstance(http_response.raw, httpx.Response):
+        streaming_cls = AioHttpxStreamingChecksumBody
+    else:
+        streaming_cls = AioStreamingChecksumBody
+    return streaming_cls(
+        http_response.raw,
+        response["headers"].get("content-length"),
+        checksum_cls(),
+        response["headers"][header_name],
+    )
 
 
 async def handle_checksum_body(
@@ -200,21 +192,6 @@ async def handle_checksum_body(
     )
 
 
-def _handle_streaming_response(http_response, response, algorithm):
-    checksum_cls = _CHECKSUM_CLS.get(algorithm)
-    header_name = f"x-amz-checksum-{algorithm}"
-    if httpx is not None and isinstance(http_response.raw, httpx.Response):
-        streaming_cls = HttpxStreamingChecksumBody
-    else:
-        streaming_cls = StreamingChecksumBody
-    return streaming_cls(
-        http_response.raw,
-        response["headers"].get("content-length"),
-        checksum_cls(),
-        response["headers"][header_name],
-    )
-
-
 async def _handle_bytes_response(http_response, response, algorithm):
     body = await http_response.content
     header_name = f"x-amz-checksum-{algorithm}"