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/13579.deprecation.rst

@@ -0,0 +1,2 @@
+Deprecated ``ClientResponse.output_size`` and ``ClientResponse.upload_complete``;
+use ``aiohttp.UploadTracker`` instead -- by :user:`Dreamsorcerer`.

CHANGES/13579.feature.rst

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

THREAT_MODEL.md

@@ -31,7 +31,7 @@ Key public APIs (non-exhaustive):
 | Surface | Entry points |
 | --- | --- |
 | Server | `aiohttp.web.Application`, `web.RouteTableDef`, `web.run_app`, `web.AppRunner`, `web.WebSocketResponse`, `web.FileResponse` |
-| Client | `aiohttp.ClientSession`, `aiohttp.TCPConnector`, `aiohttp.ClientResponse`, `aiohttp.WSMessage`, `aiohttp.BasicAuth` |
+| Client | `aiohttp.ClientSession`, `aiohttp.TCPConnector`, `aiohttp.ClientResponse`, `aiohttp.UploadTracker`, `aiohttp.WSMessage`, `aiohttp.BasicAuth` |
 | Shared | `aiohttp.MultipartReader`/`MultipartWriter`, `aiohttp.CookieJar`, `aiohttp.TraceConfig`, `aiohttp.resolver.AsyncResolver` |
 
 ---

aiohttp/__init__.py

@@ -43,6 +43,8 @@
     TCPConnector,
     TooManyRedirects,
     UnixConnector,
+    UploadAbortedError,
+    UploadTracker,
     WSMessageTypeError,
     WSServerHandshakeError,
     request,
@@ -156,6 +158,8 @@
     "TCPConnector",
     "TooManyRedirects",
     "UnixConnector",
+    "UploadAbortedError",
+    "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

@@ -65,6 +65,7 @@
     ServerTimeoutError,
     SocketTimeoutError,
     TooManyRedirects,
+    UploadAbortedError,
     WSMessageTypeError,
     WSServerHandshakeError,
 )
@@ -77,6 +78,7 @@
     Fingerprint,
     RequestInfo,
     ResponseParams,
+    UploadTracker,
 )
 from .client_ws import (
     DEFAULT_WS_CLIENT_TIMEOUT,
@@ -137,12 +139,14 @@
     "ServerTimeoutError",
     "SocketTimeoutError",
     "TooManyRedirects",
+    "UploadAbortedError",
     "WSServerHandshakeError",
     # client_reqrep
     "ClientRequest",
     "ClientResponse",
     "Fingerprint",
     "RequestInfo",
+    "UploadTracker",
     # connector
     "BaseConnector",
     "TCPConnector",
@@ -194,6 +198,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,111 +499,142 @@ 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
         # to use the existing timeouts by setting timeout to None.
 
-        if self.closed:
-            raise RuntimeError("Session is closed")
+        # Bound outside the settle-guaranteeing try block: a rebind error
+        # must not settle a tracker owned by another request.
+        if upload_tracker is not None:
+            upload_tracker._bind()
 
-        method = method.upper()
-
-        if ssl is sentinel:
-            ssl = self._default_ssl
-        if not isinstance(ssl, SSL_ALLOWED_TYPES):
-            raise TypeError(
-                "ssl should be SSLContext, Fingerprint, or bool, "
-                f"got {ssl!r} instead."
-            )
+        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
+        traces: list[Trace] = []
+        traces_started = 0
+        trace_url: URL | None = None
+        trace_headers: "CIMultiDict[str] | None" = None
 
-        if data is not None and json is not None:
-            raise ValueError(
-                "data and json parameters can not be used at the same time"
-            )
-        elif json is not None:
-            if self._json_serialize_bytes is not None:
-                data = payload.JsonBytesPayload(json, dumps=self._json_serialize_bytes)
-            else:
-                data = payload.JsonPayload(json, dumps=self._json_serialize)
+        try:
+            if self.closed:
+                raise RuntimeError("Session is closed")
 
-        redirects = 0
-        history: list[ClientResponse] = []
-        version = self._version
-        params = params or {}
+            method = method.upper()
 
-        # Merge with default headers and transform to CIMultiDict
-        headers = self._prepare_headers(headers)
+            if ssl is sentinel:
+                ssl = self._default_ssl
+            if not isinstance(ssl, SSL_ALLOWED_TYPES):
+                raise TypeError(
+                    "ssl should be SSLContext, Fingerprint, or bool, "
+                    f"got {ssl!r} instead."
+                )
 
-        try:
-            url = self._build_url(str_or_url)
-        except ValueError as e:
-            raise InvalidUrlClientError(str_or_url) from e
+            if data is not None and json is not None:
+                raise ValueError(
+                    "data and json parameters can not be used at the same time"
+                )
+            elif json is not None:
+                if self._json_serialize_bytes is not None:
+                    data = payload.JsonBytesPayload(
+                        json, dumps=self._json_serialize_bytes
+                    )
+                else:
+                    data = payload.JsonPayload(json, dumps=self._json_serialize)
 
-        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
+            ]
+
+            trace_url = url.update_query(params)
+            trace_headers = headers
+            for trace in traces:
+                await trace.send_request_start(method, trace_url, trace_headers)
+                traces_started += 1
+        except BaseException as e:
+            tm.close()
+            if handle is not None:
+                handle.cancel()
+            if upload_tracker is not None:
+                upload_tracker._finalize()
+            # Traces that saw send_request_start must also see a terminal
+            # event; traces whose start never ran get neither.
+            for trace in traces[:traces_started]:
+                assert trace_url is not None and trace_headers is not None
+                await trace.send_request_exception(method, trace_url, trace_headers, e)
+            raise
 
         timer = tm.timer()
         req: ClientRequest | None = None
@@ -709,6 +745,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 +920,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()
             return resp
 
         except BaseException as e:
@@ -892,6 +931,9 @@ async def _request(
                 handle.cancel()
                 handle = None
 
+            if upload_tracker is not None:
+                upload_tracker._finalize()
+
             if req is not None and req._body is not None:
                 await req._body.close()
 

aiohttp/client_exceptions.py

@@ -41,6 +41,7 @@
     "WSServerHandshakeError",
     "ContentTypeError",
     "ClientPayloadError",
+    "UploadAbortedError",
     "InvalidURL",
     "InvalidUrlClientError",
     "RedirectClientError",
@@ -261,6 +262,10 @@ class ClientPayloadError(ClientError):
     """Response payload error."""
 
 
+class UploadAbortedError(ClientError):
+    """The request body was never fully sent."""
+
+
 class InvalidURL(ClientError, ValueError):
     """Invalid URL.