UploadTracker

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,2 @@
+Deprecated ``ClientResponse.output_size`` and ``ClientResponse.upload_complete``;
+use ``aiohttp.UploadTracker`` instead -- by :user:`Dreamsorcerer`.

CHANGES/13427.feature.rst

@@ -0,0 +1 @@
+Added :class:`aiohttp.UploadTracker` for observing a client request's upload progress -- by :user:`Dreamsorcerer`.

aiohttp/__init__.py

@@ -43,6 +43,7 @@
     TCPConnector,
     TooManyRedirects,
     UnixConnector,
+    UploadTracker,
     WSMessageTypeError,
     WSServerHandshakeError,
     request,
@@ -156,6 +157,7 @@
     "TCPConnector",
     "TooManyRedirects",
     "UnixConnector",
+    "UploadTracker",
     "NamedPipeConnector",
     "WSServerHandshakeError",
     "request",

aiohttp/abc.py

@@ -204,6 +204,12 @@ class AbstractStreamWriter(ABC):
     buffer_size: int = 0
     output_size: int = 0
     length: int | None = 0
+    # Called with each accepted body chunk's byte count (before any
+    # transport-level transformation such as compression or chunked
+    # framing). Assigned by the client request machinery for upload
+    # progress tracking; write()/write_eof() implementations should
+    # invoke it for every body chunk they accept.
+    on_body_write: Callable[[int], None] | None = None
 
     @abstractmethod
     async def write(

aiohttp/client.py

@@ -77,6 +77,7 @@
     Fingerprint,
     RequestInfo,
     ResponseParams,
+    UploadTracker,
 )
 from .client_ws import (
     DEFAULT_WS_CLIENT_TIMEOUT,
@@ -143,6 +144,7 @@
     "ClientResponse",
     "Fingerprint",
     "RequestInfo",
+    "UploadTracker",
     # connector
     "BaseConnector",
     "TCPConnector",
@@ -194,6 +196,7 @@ class _RequestOptions(TypedDict, total=False):
     max_field_size: int | None
     max_headers: int | None
     middlewares: Sequence[ClientMiddlewareType] | None
+    upload_tracker: UploadTracker | None
 
 
 class _WSConnectOptions(TypedDict, total=False):
@@ -494,6 +497,7 @@ async def _request(
         max_field_size: int | None = None,
         max_headers: int | None = None,
         middlewares: Sequence[ClientMiddlewareType] | None = None,
+        upload_tracker: UploadTracker | None = None,
     ) -> ClientResponse:
         # NOTE: timeout clamps existing connect and read timeouts.  We cannot
         # set the default to None because we need to detect if the user wants
@@ -522,83 +526,99 @@ async def _request(
             else:
                 data = payload.JsonPayload(json, dumps=self._json_serialize)
 
-        redirects = 0
-        history: list[ClientResponse] = []
-        version = self._version
-        params = params or {}
+        if upload_tracker is not None:
+            upload_tracker._bind()
 
-        # Merge with default headers and transform to CIMultiDict
-        headers = self._prepare_headers(headers)
+        real_timeout = (
+            self._timeout if timeout is sentinel or timeout is None else 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: asyncio.TimerHandle | None = None
 
         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)
+            redirects = 0
+            history: list[ClientResponse] = []
+            version = self._version
+            params = params or {}
 
-        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
+            # Merge with default headers and transform to CIMultiDict
+            headers = self._prepare_headers(headers)
 
-        if proxy is None:
-            proxy = self._default_proxy
-
-        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 auto_decompress is None:
-            auto_decompress = self._auto_decompress
+            handle = tm.start()
 
-        if max_line_size is None:
-            max_line_size = self._max_line_size
+            if read_bufsize is None:
+                read_bufsize = self._read_bufsize
 
-        if max_field_size is None:
-            max_field_size = self._max_field_size
+            if auto_decompress is None:
+                auto_decompress = self._auto_decompress
 
-        if max_headers is None:
-            max_headers = self._max_headers
+            if max_line_size is None:
+                max_line_size = self._max_line_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_field_size is None:
+                max_field_size = self._max_field_size
 
-        for trace in traces:
-            await trace.send_request_start(method, url.update_query(params), headers)
+            if max_headers is None:
+                max_headers = self._max_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:
+            tm.close()
+            if handle is not None:
+                handle.cancel()
+            if upload_tracker is not None:
+                upload_tracker._finalize(e)
+            raise
 
         timer = tm.timer()
         req: ClientRequest | None = None
@@ -709,6 +729,7 @@ async def _request(
                         traces=traces,
                         trust_env=self.trust_env,
                     )
+                    req._upload_tracker = upload_tracker
 
                     # Apply middleware (if any) - per-request middleware overrides session middleware
                     effective_middlewares = (
@@ -883,6 +904,8 @@ async def _request(
                 await trace.send_request_end(
                     method, url.update_query(params), headers, resp
                 )
+            if upload_tracker is not None:
+                upload_tracker._finalize(None)
             return resp
 
         except BaseException as e:
@@ -892,6 +915,9 @@ async def _request(
                 handle.cancel()
                 handle = None
 
+            if upload_tracker is not None:
+                upload_tracker._finalize(e)
+
             if req is not None and req._body is not None:
                 await req._body.close()