Replace broken upload attributes with new ones

CHANGES.rst

@@ -472,8 +472,7 @@ Features
 
 
 
-- Added :attr:`~aiohttp.ClientResponse.output_size` and
-  :attr:`~aiohttp.ClientResponse.upload_complete` -- by :user:`Dreamsorcerer`.
+- Added ``ClientResponse.output_size`` and ``ClientResponse.upload_complete`` -- by :user:`Dreamsorcerer`.
 
 
   *Related issues and pull requests on GitHub:*

CHANGES/13427.deprecation.rst

@@ -0,0 +1,3 @@
+Deprecated ``ClientResponse.output_size`` and ``ClientResponse.upload_complete``;
+use ``Payload.bytes_written`` and ``Payload.upload_complete`` instead
+-- by :user:`Dreamsorcerer`.

CHANGES/13427.feature.rst

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

aiohttp/client.py

@@ -522,83 +522,98 @@ async def _request(
             else:
                 data = payload.JsonPayload(json, dumps=self._json_serialize)
 
-        redirects = 0
-        history: list[ClientResponse] = []
-        version = self._version
-        params = params or {}
-
-        # Merge with default headers and transform to CIMultiDict
-        headers = self._prepare_headers(headers)
+        if isinstance(data, payload.Payload):
+            # Reused payloads carry a previous attempt's upload state; drop
+            # it before this attempt so failures raised prior to building
+            # the request are reported through upload_complete as well.
+            data._reset_upload()
 
         try:
-            url = self._build_url(str_or_url)
-        except ValueError as e:
-            raise InvalidUrlClientError(str_or_url) from e
-
-        assert self._connector is not None
-        if url.scheme not in self._connector.allowed_protocol_schema_set:
-            raise NonHttpUrlClientError(url)
-
-        skip_headers: Iterable[istr] | None
-        if skip_auto_headers is not None:
-            skip_headers = {
-                istr(i) for i in skip_auto_headers
-            } | self._skip_auto_headers
-        elif self._skip_auto_headers:
-            skip_headers = self._skip_auto_headers
-        else:
-            skip_headers = None
+            redirects = 0
+            history: list[ClientResponse] = []
+            version = self._version
+            params = params or {}
 
-        if proxy is None:
-            proxy = self._default_proxy
+            # Merge with default headers and transform to CIMultiDict
+            headers = self._prepare_headers(headers)
 
-        resolved_proxy_headers: CIMultiDict[str] | None
-        if proxy is None:
-            resolved_proxy_headers = None
-        else:
-            resolved_proxy_headers = self._prepare_headers(proxy_headers)
             try:
-                proxy = URL(proxy)
+                url = self._build_url(str_or_url)
             except ValueError as e:
-                raise InvalidURL(proxy) from e
+                raise InvalidUrlClientError(str_or_url) from e
+
+            assert self._connector is not None
+            if url.scheme not in self._connector.allowed_protocol_schema_set:
+                raise NonHttpUrlClientError(url)
+
+            skip_headers: Iterable[istr] | None
+            if skip_auto_headers is not None:
+                skip_headers = {
+                    istr(i) for i in skip_auto_headers
+                } | self._skip_auto_headers
+            elif self._skip_auto_headers:
+                skip_headers = self._skip_auto_headers
+            else:
+                skip_headers = None
 
-        if timeout is sentinel or timeout is None:
-            real_timeout: ClientTimeout = self._timeout
-        else:
-            real_timeout = timeout
-        # timeout is cumulative for all request operations
-        # (request, redirects, responses, data consuming)
-        tm = TimeoutHandle(
-            self._loop, real_timeout.total, ceil_threshold=real_timeout.ceil_threshold
-        )
-        handle = tm.start()
+            if proxy is None:
+                proxy = self._default_proxy
 
-        if read_bufsize is None:
-            read_bufsize = self._read_bufsize
+            resolved_proxy_headers: CIMultiDict[str] | None
+            if proxy is None:
+                resolved_proxy_headers = None
+            else:
+                resolved_proxy_headers = self._prepare_headers(proxy_headers)
+                try:
+                    proxy = URL(proxy)
+                except ValueError as e:
+                    raise InvalidURL(proxy) from e
+
+            if timeout is sentinel or timeout is None:
+                real_timeout: ClientTimeout = self._timeout
+            else:
+                real_timeout = timeout
+            # timeout is cumulative for all request operations
+            # (request, redirects, responses, data consuming)
+            tm = TimeoutHandle(
+                self._loop,
+                real_timeout.total,
+                ceil_threshold=real_timeout.ceil_threshold,
+            )
+            handle = tm.start()
 
-        if auto_decompress is None:
-            auto_decompress = self._auto_decompress
+            if read_bufsize is None:
+                read_bufsize = self._read_bufsize
 
-        if max_line_size is None:
-            max_line_size = self._max_line_size
+            if auto_decompress is None:
+                auto_decompress = self._auto_decompress
 
-        if max_field_size is None:
-            max_field_size = self._max_field_size
+            if max_line_size is None:
+                max_line_size = self._max_line_size
 
-        if max_headers is None:
-            max_headers = self._max_headers
+            if max_field_size is None:
+                max_field_size = self._max_field_size
 
-        traces = [
-            Trace(
-                self,
-                trace_config,
-                trace_config.trace_config_ctx(trace_request_ctx=trace_request_ctx),
-            )
-            for trace_config in self._trace_configs
-        ]
+            if max_headers is None:
+                max_headers = self._max_headers
 
-        for trace in traces:
-            await trace.send_request_start(method, url.update_query(params), headers)
+            traces = [
+                Trace(
+                    self,
+                    trace_config,
+                    trace_config.trace_config_ctx(trace_request_ctx=trace_request_ctx),
+                )
+                for trace_config in self._trace_configs
+            ]
+
+            for trace in traces:
+                await trace.send_request_start(
+                    method, url.update_query(params), headers
+                )
+        except BaseException as e:
+            if isinstance(data, payload.Payload) and not data._upload_active:
+                data._abort_upload(e)
+            raise
 
         timer = tm.timer()
         req: ClientRequest | None = None
@@ -895,6 +910,14 @@ async def _request(
             if req is not None and req._body is not None:
                 await req._body.close()
 
+            # Ensure we abort when the write never started.
+            if req is not None:
+                body = None if req._body is req._EMPTY_BODY else req._body
+            else:
+                body = data if isinstance(data, payload.Payload) else None
+            if body is not None and not body._upload_active:
+                body._abort_upload(e)
+
             for trace in traces:
                 await trace.send_request_exception(
                     method, url.update_query(params), headers, e

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
@@ -1266,6 +1288,10 @@ def _update_body_from_data(self, body: Any) -> None:
                 body = FormData(body, boundary=boundary)()
 
         self._body = body
+        # A payload may be reused from an earlier request (redirects,
+        # retries, or explicit reuse): drop that attempt's upload state so
+        # a failure of this request cannot be masked by a stale outcome.
+        body._reset_upload()
 
         # enable chunked encoding if needed
         if not self.chunked and hdrs.CONTENT_LENGTH not in self.headers:
@@ -1447,6 +1473,11 @@ 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:
+        body = self._body
+        if body is not self._EMPTY_BODY and body._start_upload():
+            body._finish_upload()
+
     async def _write_bytes(
         self,
         writer: AbstractStreamWriter,
@@ -1472,55 +1503,71 @@ 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
+        # Progress tracking is exclusive per payload; when overlapping
+        # requests share one instance, later uploads run untracked.
+        track_progress = body is not self._EMPTY_BODY and body._start_upload()
+        if track_progress:
+            writer = payload._ProgressWriter(writer, body)
+        abort_exc: BaseException | None = None
         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
                 )
-
-            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(
+                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)
+                abort_exc = reraised_exc
+            except asyncio.CancelledError:
+                # Body hasn't been fully sent, so connection can't be reused
+                conn.close()
+                raise
+            except Exception as underlying_exc:
+                abort_exc = 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, abort_exc, underlying_exc)
+            else:
+                # Successfully wrote the body, signal EOF and start response timeout
+                await writer.write_eof()
+                if track_progress:
+                    body._finish_upload()
+                protocol.start_timeout()
+        finally:
+            if track_progress:
+                # No-op when the upload finished; reports the write error
+                # otherwise, or cancels upload_complete when there is none
+                # (cancellation or a failure that bypassed the handlers).
+                body._abort_upload(abort_exc)
 
     async def _close(self) -> None:
         if self._writer_task is not None: