[PR #11011/8658faad backport][3.12] Correct type of ClientRequest.body (#11023)

Co-authored-by: Sam Bull <[email protected]>
Co-authored-by: J. Nick Koston <[email protected]>

Author: 3 people

aiohttp/client_middleware_digest_auth.py

@@ -193,7 +193,9 @@ def __init__(
         self._nonce_count = 0
         self._challenge: DigestAuthChallenge = {}
 
-    async def _encode(self, method: str, url: URL, body: Union[bytes, Payload]) -> str:
+    async def _encode(
+        self, method: str, url: URL, body: Union[Payload, Literal[b""]]
+    ) -> str:
         """
         Build digest authorization header for the current challenge.
 
@@ -274,10 +276,10 @@ def KD(s: bytes, d: bytes) -> bytes:
         A1 = b":".join((self._login_bytes, realm_bytes, self._password_bytes))
         A2 = f"{method.upper()}:{path}".encode()
         if qop == "auth-int":
-            if isinstance(body, bytes):  # will always be empty bytes unless Payload
-                entity_bytes = body
-            else:
+            if isinstance(body, Payload):  # will always be empty bytes unless Payload
                 entity_bytes = await body.as_bytes()  # Get bytes from Payload
+            else:
+                entity_bytes = body
             entity_hash = H(entity_bytes)
             A2 = b":".join((A2, entity_hash))
 

aiohttp/client_reqrep.py

@@ -17,6 +17,7 @@
     Dict,
     Iterable,
     List,
+    Literal,
     Mapping,
     NamedTuple,
     Optional,
@@ -459,7 +460,7 @@ def port(self) -> Optional[int]:
         return self.url.port
 
     @property
-    def body(self) -> Union[bytes, payload.Payload]:
+    def body(self) -> Union[payload.Payload, Literal[b""]]:
         """Request body."""
         # empty body is represented as bytes for backwards compatibility
         return self._body or b""

docs/client_reference.rst

@@ -1882,12 +1882,9 @@ ClientRequest
    For more information about using middleware, see :ref:`aiohttp-client-middleware`.
 
    .. attribute:: body
-      :type: Payload | FormData
+      :type: Payload | Literal[b""]
 
-      The request body payload. This can be:
-
-      - A :class:`Payload` object for raw data (default is empty bytes ``b""``)
-      - A :class:`FormData` object for form submissions
+      The request body payload (defaults to ``b""`` if no body passed).
 
       .. danger::
 

tests/test_client_middleware_digest_auth.py

@@ -2,7 +2,7 @@
 
 import io
 from hashlib import md5, sha1
-from typing import Generator, Union
+from typing import Generator, Literal, Union
 from unittest import mock
 
 import pytest
@@ -270,7 +270,7 @@ def KD(secret: str, data: str) -> str:
 @pytest.mark.parametrize(
     ("body", "body_str"),
     [
-        (b"this is a body", "this is a body"),  # Bytes case
+        (b"", ""),  # Bytes case
         (
             BytesIOPayload(io.BytesIO(b"this is a body")),
             "this is a body",
@@ -280,7 +280,7 @@ def KD(secret: str, data: str) -> str:
 async def test_digest_response_exact_match(
     qop: str,
     algorithm: str,
-    body: Union[bytes, BytesIOPayload],
+    body: Union[Literal[b""], BytesIOPayload],
     body_str: str,
     mock_sha1_digest: mock.MagicMock,
 ) -> None:

tests/test_client_request.py

@@ -1261,7 +1261,7 @@ def read(self, decode=False):
 
 async def test_oserror_on_write_bytes(loop, conn) -> None:
     req = ClientRequest("POST", URL("http://python.org/"), loop=loop)
-    req.body = b"test data"
+    req.body = b"test data"  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
 
     writer = WriterMock()
     writer.write.side_effect = OSError
@@ -1618,7 +1618,7 @@ async def test_write_bytes_with_content_length_limit(
     data = b"Hello World"
     req = ClientRequest("post", URL("http://python.org/"), loop=loop)
 
-    req.body = data
+    req.body = data  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
 
     writer = StreamWriter(protocol=conn.protocol, loop=loop)
     # Use content_length=5 to truncate data
@@ -1655,7 +1655,7 @@ async def gen() -> AsyncIterator[bytes]:
 
         req.body = gen()  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
     else:
-        req.body = data
+        req.body = data  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
 
     writer = StreamWriter(protocol=conn.protocol, loop=loop)
     # Use content_length=7 to truncate at the middle of Part2
@@ -1705,7 +1705,7 @@ async def test_warn_if_unclosed_payload_via_body_setter(
         ResourceWarning,
         match="The previous request body contains unclosed resources",
     ):
-        req.body = b"new data"
+        req.body = b"new data"  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
 
     await req.close()
 
@@ -1723,7 +1723,7 @@ async def test_no_warn_for_autoclose_payload_via_body_setter(
     # Setting body again should not trigger warning since previous payload has autoclose=True
     with warnings.catch_warnings(record=True) as warning_list:
         warnings.simplefilter("always")
-        req.body = b"new data"
+        req.body = b"new data"  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
 
     # Filter out any non-ResourceWarning warnings
     resource_warnings = [
@@ -1753,7 +1753,7 @@ async def test_no_warn_for_consumed_payload_via_body_setter(
     # Setting body again should not trigger warning since previous payload is consumed
     with warnings.catch_warnings(record=True) as warning_list:
         warnings.simplefilter("always")
-        req.body = b"new data"
+        req.body = b"new data"  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
 
     # Filter out any non-ResourceWarning warnings
     resource_warnings = [
@@ -1872,7 +1872,7 @@ async def test_body_setter_closes_previous_payload(
     req._body = mock_payload
 
     # Update body with new data using setter
-    req.body = b"new body data"
+    req.body = b"new body data"  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892
 
     # Verify the previous payload was closed using _close
     mock_payload._close.assert_called_once()
@@ -2001,7 +2001,7 @@ async def test_warn_stacklevel_points_to_user_code(
     with warnings.catch_warnings(record=True) as warning_list:
         warnings.simplefilter("always", ResourceWarning)
         # This line should be reported as the warning source
-        req.body = b"new data"  # LINE TO BE REPORTED
+        req.body = b"new data"  # type: ignore[assignment]  # https://github.com/python/mypy/issues/12892  # LINE TO BE REPORTED
 
     # Find the ResourceWarning
     resource_warnings = [