Replace broken upload attributes with new ones

CHANGES/13427.deprecation.rst

@@ -0,0 +1,4 @@
+Deprecated ``ClientResponse.output_size`` and ``ClientResponse.upload_complete``;
+use :attr:`Payload.bytes_written <aiohttp.Payload.bytes_written>` and
+:attr:`Payload.upload_complete <aiohttp.Payload.upload_complete>` instead
+-- by :user:`Dreamsorcerer`.

CHANGES/13427.feature.rst

@@ -0,0 +1,4 @@
+Added :attr:`Payload.bytes_written <aiohttp.Payload.bytes_written>` and
+:attr:`Payload.upload_complete <aiohttp.Payload.upload_complete>` for tracking
+upload progress of a request body
+-- by :user:`Dreamsorcerer`.

aiohttp/client_reqrep.py

@@ -361,7 +361,17 @@ def _writer(self, writer: asyncio.Task[None] | None) -> None:
 
     @property
     def output_size(self) -> int:
-        """Number of bytes sent for this request."""
+        """Number of bytes sent for this request.
+
+        .. deprecated:: 3.14.4
+           Use :attr:`Payload.bytes_written` instead.
+        """
+        warnings.warn(
+            "ClientResponse.output_size is deprecated, "
+            "use Payload.bytes_written instead",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         if self._stream_writer is not None:
             return self._stream_writer.output_size
         return self._output_size
@@ -370,8 +380,15 @@ def output_size(self) -> int:
     def upload_complete(self) -> "asyncio.Future[None]":
         """Future set when the request body has been fully sent.
 
-        Already done when the request had no body or was written eagerly.
+        .. deprecated:: 3.14.4
+           Use :attr:`Payload.upload_complete` instead.
         """
+        warnings.warn(
+            "ClientResponse.upload_complete is deprecated, "
+            "use Payload.upload_complete instead",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         if self._upload_complete is None:
             self._upload_complete = self._loop.create_future()
             if self._stream_writer is None:  # upload already finished
@@ -1001,6 +1018,7 @@ async def _send(self, conn: "Connection") -> ClientResponse:
             protocol.start_timeout()
             writer.set_eof()
             task = None
+            self._mark_body_sent()
         self._response = self._create_response(task, stream_writer=writer)
         return self._response
 
@@ -1013,6 +1031,10 @@ async def _write_bytes(
         # Base class never has a body, this will never be run.
         assert False
 
+    def _mark_body_sent(self) -> None:
+        """Hook invoked when the request is sent without a body to write."""
+        # Base class requests (CONNECT) never carry a body payload.
+
 
 class ClientRequestArgs(TypedDict, total=False):
     params: Query
@@ -1447,6 +1469,10 @@ def _should_write(self, protocol: BaseProtocol) -> bool:
             self.body.size != 0 or self._continue is not None or protocol.writing_paused
         )
 
+    def _mark_body_sent(self) -> None:
+        if (body := self._body) is not self._EMPTY_BODY:
+            body._finish_upload()
+
     async def _write_bytes(
         self,
         writer: AbstractStreamWriter,
@@ -1472,55 +1498,68 @@ async def _write_bytes(
         - Content length constraints for chunked encoding
         - Error handling for network issues, cancellation, and other exceptions
         - Signaling EOF and timeout management
+        - Upload progress bookkeeping on the payload
+          (:attr:`Payload.bytes_written` / :attr:`Payload.upload_complete`)
 
         Raises:
             ClientOSError: When there's an OS-level error writing the body
             ClientConnectionError: When there's a general connection error
             asyncio.CancelledError: When the operation is cancelled
 
         """
-        # 100 response
-        if self._continue is not None:
-            # Force headers to be sent before waiting for 100-continue
-            writer.send_headers()
-            await writer.drain()
-            await self._continue
-
-        protocol = conn.protocol
-        assert protocol is not None
+        body = self._body
+        track_progress = body is not self._EMPTY_BODY
+        if track_progress:
+            body._start_upload()
+            writer = payload._ProgressWriter(writer, body)
         try:
-            await self._body.write_with_length(writer, content_length)
-        except OSError as underlying_exc:
-            reraised_exc = underlying_exc
+            # 100 response
+            if self._continue is not None:
+                # Force headers to be sent before waiting for 100-continue
+                writer.send_headers()
+                await writer.drain()
+                await self._continue
+
+            protocol = conn.protocol
+            assert protocol is not None
+            try:
+                await body.write_with_length(writer, content_length)
+            except OSError as underlying_exc:
+                reraised_exc = underlying_exc
 
-            # Distinguish between timeout and other OS errors for better error reporting
-            exc_is_not_timeout = underlying_exc.errno is not None or not isinstance(
-                underlying_exc, asyncio.TimeoutError
-            )
-            if exc_is_not_timeout:
-                reraised_exc = ClientOSError(
-                    underlying_exc.errno,
-                    f"Can not write request body for {self.url !s}",
+                # Distinguish between timeout and other OS errors for better error reporting
+                exc_is_not_timeout = underlying_exc.errno is not None or not isinstance(
+                    underlying_exc, asyncio.TimeoutError
                 )
+                if exc_is_not_timeout:
+                    reraised_exc = ClientOSError(
+                        underlying_exc.errno,
+                        f"Can not write request body for {self.url !s}",
+                    )
 
-            set_exception(protocol, reraised_exc, underlying_exc)
-        except asyncio.CancelledError:
-            # Body hasn't been fully sent, so connection can't be reused
-            conn.close()
-            raise
-        except Exception as underlying_exc:
-            set_exception(
-                protocol,
-                ClientConnectionError(
-                    "Failed to send bytes into the underlying connection "
-                    f"{conn !s}: {underlying_exc!r}",
-                ),
-                underlying_exc,
-            )
-        else:
-            # Successfully wrote the body, signal EOF and start response timeout
-            await writer.write_eof()
-            protocol.start_timeout()
+                set_exception(protocol, reraised_exc, underlying_exc)
+            except asyncio.CancelledError:
+                # Body hasn't been fully sent, so connection can't be reused
+                conn.close()
+                raise
+            except Exception as underlying_exc:
+                set_exception(
+                    protocol,
+                    ClientConnectionError(
+                        "Failed to send bytes into the underlying connection "
+                        f"{conn !s}: {underlying_exc!r}",
+                    ),
+                    underlying_exc,
+                )
+            else:
+                # Successfully wrote the body, signal EOF and start response timeout
+                await writer.write_eof()
+                body._finish_upload()
+                protocol.start_timeout()
+        finally:
+            if track_progress:
+                # No-op when the upload finished; cancels upload_complete otherwise.
+                body._abort_upload()
 
     async def _close(self) -> None:
         if self._writer_task is not None:

aiohttp/payload.py

@@ -142,6 +142,11 @@ class Payload(ABC):
     _size: int | None = None
     _consumed: bool = False  # Default: payload has not been consumed yet
     _autoclose: bool = False  # Default: assume resource needs explicit closing
+    # Upload progress bookkeeping (client requests); see bytes_written.
+    _bytes_written: int = 0
+    _upload_future: "asyncio.Future[None] | None" = None
+    _upload_finished: bool = False
+    _upload_aborted: bool = False
 
     def __init__(
         self,
@@ -230,6 +235,65 @@ def autoclose(self) -> bool:
         """
         return self._autoclose
 
+    @property
+    def bytes_written(self) -> int:
+        """Number of bytes of this payload written to the connection so far.
+
+        Counts the bytes handed to the transport while a client request
+        is uploading this payload, before any transport-level transformation
+        (compression, chunked framing). Reset when a new upload of the same
+        payload starts, e.g. when a redirect causes the body to be resent.
+        """
+        return self._bytes_written
+
+    @property
+    def upload_complete(self) -> "asyncio.Future[None]":
+        """Future resolved when a request finishes uploading this payload.
+
+        The future completes with ``None`` once the request body has been
+        fully written, or is cancelled if the upload is interrupted (e.g.
+        connection error or request cancellation). If the payload is sent
+        again (e.g. a redirected request resends the body), a new future
+        is returned for the new upload.
+
+        Must be accessed from within the event loop.
+        """
+        if self._upload_future is None:
+            self._upload_future = asyncio.get_running_loop().create_future()
+            if self._upload_aborted:
+                self._upload_future.cancel()
+            elif self._upload_finished:
+                self._upload_future.set_result(None)
+        return self._upload_future
+
+    def _start_upload(self) -> None:
+        """Reset upload progress state before the payload is written."""
+        self._bytes_written = 0
+        self._upload_finished = False
+        self._upload_aborted = False
+        fut = self._upload_future
+        if fut is not None and fut.done():
+            # A previous upload of this payload already completed (e.g. the
+            # request is resent after a redirect); track the new upload with
+            # a fresh future created lazily on the next upload_complete access.
+            self._upload_future = None
+
+    def _finish_upload(self) -> None:
+        """Mark the payload as fully written to the connection."""
+        self._upload_finished = True
+        fut = self._upload_future
+        if fut is not None and not fut.done():
+            fut.set_result(None)
+
+    def _abort_upload(self) -> None:
+        """Mark an unfinished upload as aborted, cancelling upload_complete."""
+        if self._upload_finished:
+            return
+        self._upload_aborted = True
+        fut = self._upload_future
+        if fut is not None and not fut.done():
+            fut.cancel()
+
     def set_content_disposition(
         self,
         disptype: str,
@@ -337,6 +401,79 @@ async def close(self) -> None:
         self._close()
 
 
+class _ProgressWriter(AbstractStreamWriter):
+    """Proxy for AbstractStreamWriter to count written bytes on a Payload.
+
+    Used by the client request machinery so that Payload.bytes_written
+    reflects how much of the payload has been handed to the transport.
+    """
+
+    def __init__(self, writer: AbstractStreamWriter, payload: "Payload") -> None:
+        self._writer = writer
+        self._payload = payload
+
+    def __getattr__(self, name: str) -> Any:
+        # Forward non-interface attributes of the wrapped writer
+        # (e.g. StreamWriter.transport) for duck-typing compatibility.
+        return getattr(self._writer, name)
+
+    @property
+    def buffer_size(self) -> int:
+        return self._writer.buffer_size
+
+    @buffer_size.setter
+    def buffer_size(self, value: int) -> None:
+        self._writer.buffer_size = value
+
+    @property
+    def output_size(self) -> int:
+        return self._writer.output_size
+
+    @output_size.setter
+    def output_size(self, value: int) -> None:
+        self._writer.output_size = value
+
+    @property
+    def length(self) -> int | None:
+        return self._writer.length
+
+    @length.setter
+    def length(self, value: int | None) -> None:
+        self._writer.length = value
+
+    async def write(
+        self, chunk: "bytes | bytearray | memoryview[int] | memoryview[bytes]"
+    ) -> None:
+        await self._writer.write(chunk)
+        self._payload._bytes_written += (
+            chunk.nbytes if isinstance(chunk, memoryview) else len(chunk)
+        )
+
+    async def write_eof(self, chunk: bytes = b"") -> None:
+        await self._writer.write_eof(chunk)
+        if chunk:
+            self._payload._bytes_written += len(chunk)
+
+    async def drain(self) -> None:
+        await self._writer.drain()
+
+    def enable_compression(
+        self, encoding: str = "deflate", strategy: int | None = None
+    ) -> None:
+        self._writer.enable_compression(encoding, strategy)
+
+    def enable_chunking(self) -> None:
+        self._writer.enable_chunking()
+
+    async def write_headers(
+        self, status_line: str, headers: "CIMultiDict[str]"
+    ) -> None:
+        await self._writer.write_headers(status_line, headers)
+
+    def send_headers(self) -> None:
+        self._writer.send_headers()
+
+
 class BytesPayload(Payload):
     _value: bytes
     # _consumed = False (inherited) - Bytes are immutable and can be reused