Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/client_reqrep.py: 32%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import asyncio
2import codecs
3import contextlib
4import functools
5import io
6import re
7import sys
8import traceback
9import warnings
10from asyncio.base_events import BaseEventLoop
11from collections import deque
12from collections.abc import Callable, Iterable, Sequence
13from hashlib import md5, sha1, sha256
14from http.cookies import BaseCookie, SimpleCookie
15from types import MappingProxyType, TracebackType
16from typing import TYPE_CHECKING, Any, Literal, NamedTuple, TypedDict
18from multidict import CIMultiDict, CIMultiDictProxy, MultiDict, MultiDictProxy
19from yarl import URL, Query
21from . import hdrs, multipart, payload
22from ._cookie_helpers import (
23 parse_cookie_header,
24 parse_set_cookie_headers,
25 preserve_morsel_with_coded_value,
26)
27from .abc import AbstractStreamWriter
28from .base_protocol import BaseProtocol
29from .client_exceptions import (
30 ClientConnectionError,
31 ClientOSError,
32 ClientResponseError,
33 ContentTypeError,
34 InvalidURL,
35 ServerFingerprintMismatch,
36 UploadAbortedError,
37)
38from .compression_utils import HAS_BROTLI, HAS_ZSTD
39from .formdata import FormData
40from .helpers import (
41 _SENTINEL,
42 HTTP_AND_EMPTY_SCHEMA_SET,
43 BaseTimerContext,
44 HeadersDictProxy,
45 HeadersMixin,
46 TimerNoop,
47 encode_basic_auth,
48 frozen_dataclass_decorator,
49 is_expected_content_type,
50 parse_mimetype,
51 reify,
52 sentinel,
53 set_exception,
54 set_result,
55)
56from .http import (
57 SERVER_SOFTWARE,
58 HttpProcessingError,
59 HttpVersion,
60 HttpVersion10,
61 HttpVersion11,
62 StreamWriter,
63)
64from .streams import EMPTY_PAYLOAD, StreamReader
65from .typedefs import DEFAULT_JSON_DECODER, JSONDecoder, RawHeaders
67try:
68 import ssl
69 from ssl import SSLContext
70except ImportError: # pragma: no cover
71 ssl = None # type: ignore[assignment]
72 SSLContext = object # type: ignore[misc,assignment]
75__all__ = ("ClientRequest", "ClientResponse", "RequestInfo", "Fingerprint")
78if TYPE_CHECKING:
79 from .client import ClientSession
80 from .connector import Connection
81 from .tracing import Trace
84_CONNECTION_CLOSED_EXCEPTION = ClientConnectionError("Connection closed")
85_CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
86_DIGITS_RE = re.compile(r"\d+", re.ASCII)
87_LINK_PARAM_RE = re.compile(r"^([^\s=]+)\s*=\s*(?:(['\"])(.*?)\2|(\S*))$", re.M)
90@frozen_dataclass_decorator
91class ClientTimeout:
92 total: float | None = 5 * 60 # 5 minute default timeout
93 connect: float | None = None
94 sock_read: float | None = None
95 sock_connect: float | None = None
96 ceil_threshold: float = 5
98 # pool_queue_timeout: Optional[float] = None
99 # dns_resolution_timeout: Optional[float] = None
100 # socket_connect_timeout: Optional[float] = None
101 # connection_acquiring_timeout: Optional[float] = None
102 # new_connection_timeout: Optional[float] = None
103 # http_header_timeout: Optional[float] = None
104 # response_body_timeout: Optional[float] = None
106 # to create a timeout specific for a single request, either
107 # - create a completely new one to overwrite the default
108 # - or use https://docs.python.org/3/library/dataclasses.html#dataclasses.replace
109 # to overwrite the defaults
111 def __post_init__(self) -> None:
112 # Ensure total is never lower than a more specific timeout, otherwise
113 # the latter would be silently capped by total and rendered useless.
114 # total=None means the user explicitly disabled the total timeout.
115 if self.total is None:
116 return
117 object.__setattr__(
118 self,
119 "total",
120 max(
121 self.total,
122 self.connect or 0,
123 self.sock_read or 0,
124 self.sock_connect or 0,
125 ),
126 )
128 if self.total == 0:
129 raise ValueError(
130 "total timeout must be a positive number or None to disable, "
131 "got 0. Using 0 to disable timeouts is no longer supported, "
132 "use None instead."
133 )
136def _gen_default_accept_encoding() -> str:
137 encodings = [
138 "gzip",
139 "deflate",
140 ]
141 if HAS_BROTLI:
142 encodings.append("br")
143 if HAS_ZSTD:
144 encodings.append("zstd")
145 return ", ".join(encodings)
148@frozen_dataclass_decorator
149class ContentDisposition:
150 type: str | None
151 parameters: "MappingProxyType[str, str]"
152 filename: str | None
155class _RequestInfo(NamedTuple):
156 url: URL
157 method: str
158 headers: "CIMultiDictProxy[str]"
159 real_url: URL
162class RequestInfo(_RequestInfo):
164 def __new__(
165 cls,
166 url: URL,
167 method: str,
168 headers: "CIMultiDictProxy[str]",
169 real_url: URL | _SENTINEL = sentinel,
170 ) -> "RequestInfo":
171 """Create a new RequestInfo instance.
173 For backwards compatibility, the real_url parameter is optional.
174 """
175 return tuple.__new__(
176 cls, (url, method, headers, url if real_url is sentinel else real_url)
177 )
180class Fingerprint:
181 HASHFUNC_BY_DIGESTLEN = {
182 16: md5,
183 20: sha1,
184 32: sha256,
185 }
187 def __init__(self, fingerprint: bytes) -> None:
188 digestlen = len(fingerprint)
189 hashfunc = self.HASHFUNC_BY_DIGESTLEN.get(digestlen)
190 if not hashfunc:
191 raise ValueError("fingerprint has invalid length")
192 elif hashfunc is md5 or hashfunc is sha1:
193 raise ValueError("md5 and sha1 are insecure and not supported. Use sha256.")
194 self._hashfunc = hashfunc
195 self._fingerprint = fingerprint
197 @property
198 def fingerprint(self) -> bytes:
199 return self._fingerprint
201 def check(self, transport: asyncio.Transport) -> None:
202 if not transport.get_extra_info("sslcontext"):
203 return
204 sslobj = transport.get_extra_info("ssl_object")
205 cert = sslobj.getpeercert(binary_form=True)
206 got = self._hashfunc(cert).digest()
207 if got != self._fingerprint:
208 host, port, *_ = transport.get_extra_info("peername")
209 raise ServerFingerprintMismatch(self._fingerprint, got, host, port)
212if ssl is not None:
213 SSL_ALLOWED_TYPES = (ssl.SSLContext, bool, Fingerprint)
214else: # pragma: no cover
215 SSL_ALLOWED_TYPES = (bool,) # type: ignore[unreachable]
218_CONNECTION_CLOSED_EXCEPTION = ClientConnectionError("Connection closed")
219_SSL_SCHEMES = frozenset(("https", "wss"))
222# ConnectionKey is a NamedTuple because it is used as a key in a dict
223# and a set in the connector. Since a NamedTuple is a tuple it uses
224# the fast native tuple __hash__ and __eq__ implementation in CPython.
225class ConnectionKey(NamedTuple):
226 # the key should contain an information about used proxy / TLS
227 # to prevent reusing wrong connections from a pool
228 host: str
229 port: int | None
230 is_ssl: bool
231 ssl: SSLContext | bool | Fingerprint
232 proxy: URL | None
233 proxy_headers_hash: int | None # hash(CIMultiDict)
234 server_hostname: str | None = None
237class ResponseParams(TypedDict):
238 timer: BaseTimerContext | None
239 skip_payload: bool
240 read_until_eof: bool
241 auto_decompress: bool
242 read_timeout: float | None
243 read_bufsize: int
244 timeout_ceil_threshold: float
245 max_line_size: int
246 max_field_size: int
247 max_headers: int
250# Unflushed checkpoints kept before forcing a refresh in _add_bytes.
251_MAX_UPLOAD_CHECKPOINTS = 256
254class UploadTracker:
255 """Tracks upload progress of a single client request.
257 Pass a fresh instance via the request's ``upload_tracker`` argument and
258 read ``bytes_written``, ``attempts`` and ``upload_complete`` while the
259 request runs.
261 Must be created inside a running event loop. A tracker observes exactly
262 one request: passing it to a second request raises :exc:`RuntimeError`.
263 """
265 def __init__(self) -> None:
266 self.attempts = 0
267 self.upload_complete: asyncio.Future[None] = (
268 asyncio.get_running_loop().create_future()
269 )
270 self._bound = False
271 # Outcome of the latest attempt: finished, failed with _exc, or
272 # neither (never sent, or cut short).
273 self._finished = False
274 self._exc: BaseException | None = None
275 # Set once the request reaches a terminal point (returned or raised);
276 # no further attempts can start after that.
277 self._final = False
278 # Set while the current attempt's body is being written.
279 self._writer: AbstractStreamWriter | None = None
280 # Payload bytes handed to the writer for the current attempt.
281 self._accepted = 0
282 # Payload bytes known to have left the transport for the kernel.
283 self._flushed = 0
284 # (payload total, wire total) pairs: the last fully flushed point
285 # first, then accepted chunks not yet out of the transport's buffer.
286 self._checkpoints: deque[tuple[int, int]] = deque()
288 @property
289 def bytes_written(self) -> int:
290 """Payload bytes of the current attempt sent to the kernel.
292 Computed from the transport's unsent buffer on access, so it
293 excludes bytes still queued in the event loop; it cannot see past
294 the kernel's own socket buffering.
295 """
296 self._refresh_flushed()
297 return self._flushed
299 def _refresh_flushed(self) -> None:
300 writer = self._writer
301 if writer is None:
302 return
303 transport = writer.transport
304 if transport is None or transport.is_closing():
305 # Torn down: bytes dropped from the buffer were never sent,
306 # so keep the last confirmed value.
307 return
308 wire_sent = writer.output_size - transport.get_write_buffer_size()
309 checkpoints = self._checkpoints
310 while len(checkpoints) > 1 and checkpoints[1][1] <= wire_sent:
311 checkpoints.popleft()
312 # A live _writer implies the sentinel pair from _attempt_writing
313 # is present, and the loop above never pops the head.
314 flushed, wire = checkpoints[0]
315 if len(checkpoints) > 1 and wire_sent > wire:
316 # Inside a partially flushed chunk: scale the payload bytes
317 # linearly over the chunk's wire bytes (exact for identity
318 # writes, an estimate under compression).
319 payload_cum, wire_cum = checkpoints[1]
320 flushed += (payload_cum - flushed) * (wire_sent - wire) // (wire_cum - wire)
321 self._flushed = flushed
323 def _bind(self) -> None:
324 if self._bound:
325 raise RuntimeError("UploadTracker is already bound to a request")
326 self._bound = True
328 def _attempt_started(self) -> int:
329 """A new upload attempt is dispatched; returns its generation token.
331 Called before connecting, so a resend failing early settles as
332 aborted instead of reporting the superseded attempt's outcome.
333 """
334 self.attempts += 1
335 self._writer = None
336 self._accepted = 0
337 self._flushed = 0
338 self._checkpoints.clear()
339 self._finished = False
340 self._exc = None
341 return self.attempts
343 def _attempt_writing(self, gen: int, writer: AbstractStreamWriter) -> None:
344 """The attempt's body write has begun; settling now defers to it."""
345 if gen == self.attempts:
346 self._writer = writer
347 # Baseline past any bytes already on the wire (e.g. headers
348 # sent for the 100-continue preamble).
349 self._checkpoints.append((0, writer.output_size))
351 def _add_bytes(self, gen: int, size: int, wire_position: int) -> None:
352 # A stale writer of a superseded attempt (e.g. still being torn
353 # down while a redirect resends the body) must not corrupt the counter.
354 if gen == self.attempts and self._writer is not None:
355 self._accepted += size
356 checkpoints = self._checkpoints
357 # A compressor may swallow a chunk without producing output;
358 # the wire position is then unchanged and no checkpoint is due.
359 if wire_position > checkpoints[-1][1]:
360 checkpoints.append((self._accepted, wire_position))
361 # Bound the queue when nobody polls bytes_written: flushed
362 # entries are only dropped on refresh.
363 if len(checkpoints) > _MAX_UPLOAD_CHECKPOINTS:
364 self._refresh_flushed()
366 def _attempt_finished(self, gen: int) -> None:
367 if gen == self.attempts:
368 self._finished = True
369 # The body was written in full; the tail still in transit is
370 # ordered ahead of the response that completes the request.
371 self._flushed = self._accepted
372 self._checkpoints.clear()
373 self._writer = None
374 if self._final:
375 self._settle()
377 def _attempt_failed(self, gen: int, exc: BaseException | None) -> None:
378 """The attempt did not send the body in full.
380 ``exc`` is the upload error, or ``None`` when the attempt was cut
381 short (cancelled) rather than failed.
382 """
383 if gen == self.attempts:
384 self._exc = exc
385 # Keep the last kernel-confirmed value and drop the writer.
386 self._refresh_flushed()
387 self._checkpoints.clear()
388 self._writer = None
389 if self._final:
390 self._settle()
392 def _finalize(self) -> None:
393 """Mark the request terminal: no further attempts will start.
395 Settles upload_complete unless the final attempt is still writing,
396 in which case the attempt's own terminal event settles it.
397 """
398 self._final = True
399 if self._writer is None:
400 self._settle()
402 def _settle(self) -> None:
403 fut = self.upload_complete
404 if fut.done():
405 return
406 if self._finished:
407 fut.set_result(None)
408 return
409 if self._exc is not None:
410 fut.set_exception(self._exc)
411 else:
412 fut.set_exception(UploadAbortedError("The request body was not fully sent"))
413 # Avoid 'exception was never retrieved' when doing a .done() poll.
414 fut.exception()
417class ClientResponse(HeadersMixin):
418 # Some of these attributes are None when created,
419 # but will be set by the start() method.
420 # As the end user will likely never see the None values, we cheat the types below.
421 # from the Status-Line of the response
422 version: HttpVersion | None = None # HTTP-Version
423 status: int = None # type: ignore[assignment] # Status-Code
424 reason: str | None = None # Reason-Phrase
426 content: StreamReader = None # type: ignore[assignment] # Payload stream
427 _body: bytes | None = None
428 _headers: HeadersDictProxy = None # type: ignore[assignment]
429 _history: tuple["ClientResponse", ...] = ()
430 _raw_headers: RawHeaders = None # type: ignore[assignment]
431 _upgraded: bool = False # parser saw a Connection: upgrade token
433 _connection: "Connection | None" = None # current connection
434 _cookies: SimpleCookie | None = None
435 _raw_cookie_headers: tuple[str, ...] | None = None
436 _continue: asyncio.Future[bool] | None = None
437 _source_traceback: traceback.StackSummary | None = None
438 _session: "ClientSession | None" = None
439 # set up by ClientRequest after ClientResponse object creation
440 # post-init stage allows to not change ctor signature
441 _closed = True # to allow __del__ for non-initialized properly response
442 _released = False
443 _in_context = False
445 _resolve_charset: Callable[["ClientResponse", bytes], str] = lambda *_: "utf-8"
447 __writer: asyncio.Task[None] | None = None
448 _stream_writer: AbstractStreamWriter | None = None
449 _output_size: int = 0
450 _upload_complete: asyncio.Future[None] | None = None
452 def __init__(
453 self,
454 method: str,
455 url: URL,
456 *,
457 writer: asyncio.Task[None] | None,
458 continue100: asyncio.Future[bool] | None,
459 timer: BaseTimerContext | None,
460 traces: Sequence["Trace"],
461 loop: asyncio.AbstractEventLoop,
462 session: "ClientSession | None",
463 request_headers: CIMultiDict[str],
464 original_url: URL,
465 stream_writer: AbstractStreamWriter,
466 **kwargs: object,
467 ) -> None:
468 # kwargs exists so authors of subclasses should expect to pass through unknown
469 # arguments. This allows us to safely add new arguments in future releases.
470 # But, we should never receive unknown arguments here in the parent class, this
471 # would indicate an argument has been named wrong or similar in the subclass.
472 assert not kwargs, "Unexpected arguments to ClientResponse"
473 # URL forbids subclasses, so a simple type check is enough.
474 assert type(url) is URL
476 self.method = method
478 self._real_url = url
479 self._url = url.with_fragment(None) if url.raw_fragment else url
480 if writer is None: # Request already sent
481 self._output_size = stream_writer.output_size
482 else:
483 self._stream_writer = stream_writer
484 self._writer = writer
485 if continue100 is not None:
486 self._continue = continue100
487 self._request_headers = request_headers
488 self._original_url = original_url
489 self._timer = timer if timer is not None else TimerNoop()
490 self._cache: dict[str, Any] = {}
491 self._traces = traces
492 self._loop = loop
493 # Save reference to _resolve_charset, so that get_encoding() will still
494 # work after the response has finished reading the body.
495 if session is not None:
496 # store a reference to session #1985
497 self._session = session
498 self._resolve_charset = session._resolve_charset
499 if loop.get_debug():
500 self._source_traceback = traceback.extract_stack(sys._getframe(1))
502 def __reset_writer(self, _: object = None) -> None:
503 self.__writer = None
504 if self._stream_writer is not None:
505 self._output_size = self._stream_writer.output_size
506 self._stream_writer = None
507 if self._upload_complete is not None and not self._upload_complete.done():
508 self._upload_complete.set_result(None)
510 @property
511 def _writer(self) -> asyncio.Task[None] | None:
512 """The writer task for streaming data.
514 _writer is only provided for backwards compatibility
515 for subclasses that may need to access it.
516 """
517 return self.__writer
519 @_writer.setter
520 def _writer(self, writer: asyncio.Task[None] | None) -> None:
521 """Set the writer task for streaming data."""
522 if self.__writer is not None:
523 self.__writer.remove_done_callback(self.__reset_writer)
524 self.__writer = writer
525 if writer is None:
526 return
527 if writer.done():
528 # The writer is already done, so we can clear it immediately.
529 self.__reset_writer()
530 else:
531 writer.add_done_callback(self.__reset_writer)
533 @property
534 def output_size(self) -> int:
535 """Number of bytes sent for this request.
537 .. deprecated:: 3.14.4
538 Use :class:`UploadTracker` instead.
539 """
540 warnings.warn(
541 "ClientResponse.output_size is deprecated, "
542 "use aiohttp.UploadTracker instead",
543 DeprecationWarning,
544 stacklevel=2,
545 )
546 if self._stream_writer is not None:
547 return self._stream_writer.output_size
548 return self._output_size
550 @property
551 def upload_complete(self) -> "asyncio.Future[None]":
552 """Future set when the request body has been fully sent.
554 .. deprecated:: 3.14.4
555 Use :class:`UploadTracker` instead.
556 """
557 warnings.warn(
558 "ClientResponse.upload_complete is deprecated, "
559 "use aiohttp.UploadTracker instead",
560 DeprecationWarning,
561 stacklevel=2,
562 )
563 if self._upload_complete is None:
564 self._upload_complete = self._loop.create_future()
565 if self._stream_writer is None: # upload already finished
566 self._upload_complete.set_result(None)
567 return self._upload_complete
569 @property
570 def cookies(self) -> SimpleCookie:
571 if self._cookies is None:
572 if self._raw_cookie_headers is not None:
573 # Parse cookies for response.cookies (SimpleCookie for backward compatibility)
574 cookies = SimpleCookie()
575 # Use parse_set_cookie_headers for more lenient parsing that handles
576 # malformed cookies better than SimpleCookie.load
577 cookies.update(parse_set_cookie_headers(self._raw_cookie_headers))
578 self._cookies = cookies
579 else:
580 self._cookies = SimpleCookie()
581 return self._cookies
583 @cookies.setter
584 def cookies(self, cookies: SimpleCookie) -> None:
585 self._cookies = cookies
586 # Generate raw cookie headers from the SimpleCookie
587 if cookies:
588 self._raw_cookie_headers = tuple(
589 morsel.OutputString() for morsel in cookies.values()
590 )
591 else:
592 self._raw_cookie_headers = None
594 @reify
595 def url(self) -> URL:
596 return self._url
598 @reify
599 def real_url(self) -> URL:
600 return self._real_url
602 @reify
603 def host(self) -> str:
604 assert self._url.host is not None
605 return self._url.host
607 @reify
608 def headers(self) -> HeadersDictProxy:
609 return self._headers
611 @reify
612 def raw_headers(self) -> RawHeaders:
613 return self._raw_headers
615 @reify
616 def request_info(self) -> RequestInfo:
617 # Build RequestInfo lazily from components
618 headers = CIMultiDictProxy(self._request_headers)
619 return tuple.__new__(
620 RequestInfo, (self._url, self.method, headers, self._original_url)
621 )
623 @reify
624 def content_disposition(self) -> ContentDisposition | None:
625 raw = self._headers.get(hdrs.CONTENT_DISPOSITION)
626 if raw is None:
627 return None
628 disposition_type, params_dct = multipart.parse_content_disposition(raw)
629 params = MappingProxyType(params_dct)
630 filename = multipart.content_disposition_filename(params)
631 return ContentDisposition(disposition_type, params, filename)
633 def __del__(self, _warnings: Any = warnings) -> None:
634 if self._closed:
635 return
637 if self._connection is not None:
638 self._connection.release()
639 self._cleanup_writer()
641 if self._loop.get_debug():
642 _warnings.warn(
643 f"Unclosed response {self!r}", ResourceWarning, source=self
644 )
645 context = {"client_response": self, "message": "Unclosed response"}
646 if self._source_traceback:
647 context["source_traceback"] = self._source_traceback
648 self._loop.call_exception_handler(context)
650 def __repr__(self) -> str:
651 out = io.StringIO()
652 ascii_encodable_url = str(self.url)
653 if self.reason:
654 ascii_encodable_reason = self.reason.encode(
655 "ascii", "backslashreplace"
656 ).decode("ascii")
657 else:
658 ascii_encodable_reason = "None"
659 print(
660 f"<ClientResponse({ascii_encodable_url}) [{self.status} {ascii_encodable_reason}]>",
661 file=out,
662 )
663 print(self.headers, file=out)
664 return out.getvalue()
666 @property
667 def connection(self) -> "Connection | None":
668 return self._connection
670 @reify
671 def history(self) -> tuple["ClientResponse", ...]:
672 """A sequence of responses, if redirects occurred."""
673 return self._history
675 @reify
676 def links(self) -> "MultiDictProxy[MultiDictProxy[str | URL]]":
677 links: MultiDict[MultiDictProxy[str | URL]] = MultiDict()
678 for val in self.headers.getall("link"):
679 match = re.match(r"\s*<(.*)>(.*)", val)
680 if match is None: # Malformed link
681 continue
682 url, params_str = match.groups()
683 params = params_str.split(";")[1:]
685 link: MultiDict[str | URL] = MultiDict()
687 for param in params:
688 match = _LINK_PARAM_RE.match(param.strip())
689 if match is None: # Malformed param
690 continue
691 key, _, value_quoted, value_unquoted = match.groups()
693 link.add(key, value_unquoted if value_quoted is None else value_quoted)
695 key = link.get("rel", url)
697 link.add("url", self.url.join(URL(url)))
699 links.add(str(key), MultiDictProxy(link))
701 return MultiDictProxy(links)
703 async def start(self, connection: "Connection") -> "ClientResponse":
704 """Start response processing."""
705 self._closed = False
706 self._protocol = connection.protocol
707 self._connection = connection
709 with self._timer:
710 while True:
711 # read response
712 try:
713 protocol = self._protocol
714 message, payload = await protocol.read() # type: ignore[union-attr]
715 except HttpProcessingError as exc:
716 raise ClientResponseError(
717 self.request_info,
718 self.history,
719 status=exc.code,
720 message=exc.message,
721 headers=exc.headers,
722 ) from exc
724 if message.code < 100 or message.code > 199 or message.code == 101:
725 break
727 if self._continue is not None:
728 set_result(self._continue, True)
729 self._continue = None
731 # payload eof handler
732 payload.on_eof(self._response_eof)
734 # response status
735 self.version = message.version
736 self.status = message.code
737 self.reason = message.reason
739 # headers
740 self._headers = message.headers
741 self._raw_headers = message.raw_headers
742 self._upgraded = message.upgrade
744 # payload
745 self.content = payload
747 if self._traces and payload is not EMPTY_PAYLOAD:
748 payload._on_chunk_received = self._on_chunk_response_received
750 # cookies
751 if cookie_hdrs := self.headers._md.getall(hdrs.SET_COOKIE, ()):
752 # Store raw cookie headers for CookieJar
753 self._raw_cookie_headers = tuple(cookie_hdrs)
754 return self
756 def _response_eof(self) -> None:
757 if self._closed:
758 return
760 # protocol could be None because connection could be detached
761 protocol = self._connection and self._connection.protocol
762 if protocol is not None and protocol.upgraded:
763 return
765 self._closed = True
766 self._cleanup_writer()
767 self._release_connection()
769 @property
770 def closed(self) -> bool:
771 return self._closed
773 def close(self) -> None:
774 if not self._released:
775 self._notify_content()
777 self._closed = True
778 if self._loop.is_closed():
779 return
781 self._cleanup_writer()
782 if self._connection is not None:
783 self._connection.close()
784 self._connection = None
786 def release(self) -> None:
787 if not self._released:
788 self._notify_content()
790 self._closed = True
792 self._cleanup_writer()
793 self._release_connection()
795 @property
796 def ok(self) -> bool:
797 """Returns ``True`` if ``status`` is less than ``400``, ``False`` if not.
799 This is **not** a check for ``200 OK`` but a check that the response
800 status is under 400.
801 """
802 return 400 > self.status
804 def raise_for_status(self) -> None:
805 if not self.ok:
806 # reason should always be not None for a started response
807 assert self.reason is not None
809 # If we're in a context we can rely on __aexit__() to release as the
810 # exception propagates.
811 if not self._in_context:
812 self.release()
814 raise ClientResponseError(
815 self.request_info,
816 self.history,
817 status=self.status,
818 message=self.reason,
819 headers=self.headers,
820 )
822 def _release_connection(self) -> None:
823 if self._connection is not None:
824 if self.__writer is None:
825 self._connection.release()
826 self._connection = None
827 else:
828 self.__writer.add_done_callback(lambda f: self._release_connection())
830 async def _wait_released(self) -> None:
831 if self.__writer is not None:
832 try:
833 await self.__writer
834 except asyncio.CancelledError:
835 if (task := asyncio.current_task()) and task.cancelling():
836 raise
837 self._release_connection()
839 def _cleanup_writer(self) -> None:
840 if self.__writer is not None:
841 self.__writer.cancel()
842 if self._stream_writer is not None:
843 self._output_size = self._stream_writer.output_size
844 self._stream_writer = None
845 self._session = None
847 def _notify_content(self) -> None:
848 content = self.content
849 # content can be None here, but the types are cheated elsewhere.
850 if content: # type: ignore[truthy-bool]
851 if content.exception() is None:
852 set_exception(content, _CONNECTION_CLOSED_EXCEPTION)
853 # The bound method installed in start() captures self, creating a
854 # response→payload→method→self cycle. Clear it eagerly so the
855 # response is reclaimable without waiting for cycle GC.
856 if content._on_chunk_received is not None:
857 content._on_chunk_received = None
858 self._released = True
860 async def wait_for_close(self) -> None:
861 if self.__writer is not None:
862 try:
863 await self.__writer
864 except asyncio.CancelledError:
865 if (task := asyncio.current_task()) and task.cancelling():
866 raise
867 self.release()
869 async def _on_chunk_response_received(self, chunk: bytes) -> None:
870 try:
871 for trace in self._traces:
872 await trace.send_response_chunk_received(self.method, self.url, chunk)
873 except BaseException:
874 self.close()
875 raise
877 async def read(self) -> bytes:
878 """Read response payload."""
879 if self._body is None:
880 try:
881 self._body = await self.content.read()
882 except BaseException:
883 self.close()
884 raise
885 elif self._released: # Response explicitly released
886 raise ClientConnectionError("Connection closed")
888 protocol = self._connection and self._connection.protocol
889 if protocol is None or not protocol.upgraded:
890 await self._wait_released() # Underlying connection released
891 return self._body
893 def get_encoding(self) -> str:
894 ctype = self.headers.get(hdrs.CONTENT_TYPE, "").lower()
895 mimetype = parse_mimetype(ctype)
897 encoding = mimetype.parameters.get("charset")
898 if encoding:
899 with contextlib.suppress(LookupError, ValueError):
900 return codecs.lookup(encoding).name
902 if mimetype.type == "application" and (
903 mimetype.subtype == "json" or mimetype.subtype == "rdap"
904 ):
905 # RFC 7159 states that the default encoding is UTF-8.
906 # RFC 7483 defines application/rdap+json
907 return "utf-8"
909 if self._body is None:
910 raise RuntimeError(
911 "Cannot compute fallback encoding of a not yet read body"
912 )
914 return self._resolve_charset(self, self._body)
916 async def text(self, encoding: str | None = None, errors: str = "strict") -> str:
917 """Read response payload and decode."""
918 await self.read()
920 if encoding is None:
921 encoding = self.get_encoding()
923 return self._body.decode(encoding, errors=errors) # type: ignore[union-attr]
925 async def json(
926 self,
927 *,
928 encoding: str | None = None,
929 loads: JSONDecoder = DEFAULT_JSON_DECODER,
930 content_type: str | None = "application/json",
931 ) -> Any:
932 """Read and decodes JSON response."""
933 await self.read()
935 if content_type:
936 if not is_expected_content_type(self.content_type, content_type):
937 raise ContentTypeError(
938 self.request_info,
939 self.history,
940 status=self.status,
941 message=(
942 "Attempt to decode JSON with "
943 "unexpected mimetype: %s" % self.content_type
944 ),
945 headers=self.headers,
946 )
948 if encoding is None:
949 encoding = self.get_encoding()
951 return loads(self._body.decode(encoding)) # type: ignore[union-attr]
953 async def __aenter__(self) -> "ClientResponse":
954 self._in_context = True
955 return self
957 async def __aexit__(
958 self,
959 exc_type: type[BaseException] | None,
960 exc_val: BaseException | None,
961 exc_tb: TracebackType | None,
962 ) -> None:
963 self._in_context = False
964 # similar to _RequestContextManager, we do not need to check
965 # for exceptions, response object can close connection
966 # if state is broken
967 self.release()
968 await self.wait_for_close()
971class ClientRequestBase:
972 """An internal class for proxy requests."""
974 POST_METHODS = {hdrs.METH_PATCH, hdrs.METH_POST, hdrs.METH_PUT}
976 proxy: URL | None = None
977 response_class = ClientResponse
978 server_hostname: str | None = None # Needed in connector.py
979 version = HttpVersion11
980 _response = None
982 # These class defaults help create_autospec() work correctly.
983 # If autospec is improved in future, maybe these can be removed.
984 url = URL()
985 method = "GET"
987 _writer_task: asyncio.Task[None] | None = None # async task for streaming data
989 _upload_tracker: UploadTracker | None = None
990 _upload_gen = 0
992 _skip_auto_headers: "CIMultiDict[None] | None" = None
994 # N.B.
995 # Adding __del__ method with self._writer closing doesn't make sense
996 # because _writer is instance method, thus it keeps a reference to self.
997 # Until writer has finished finalizer will not be called.
999 def __init__(
1000 self,
1001 method: str,
1002 url: URL,
1003 *,
1004 headers: CIMultiDict[str],
1005 loop: asyncio.AbstractEventLoop,
1006 ssl: SSLContext | bool | Fingerprint,
1007 trust_env: bool = False,
1008 ):
1009 if match := _CONTAINS_CONTROL_CHAR_RE.search(method):
1010 raise ValueError(
1011 f"Method cannot contain non-token characters {method!r} "
1012 f"(found at least {match.group()!r})"
1013 )
1014 # URL forbids subclasses, so a simple type check is enough.
1015 assert type(url) is URL, url
1016 self.original_url = url
1017 self.url = url.with_fragment(None) if url.raw_fragment else url
1018 self.method = method.upper()
1019 self.loop = loop
1020 self._ssl = ssl
1022 if loop.get_debug():
1023 self._source_traceback = traceback.extract_stack(sys._getframe(1))
1025 if not url.raw_host:
1026 raise InvalidURL(url)
1027 self._update_headers(headers)
1028 if url.raw_user or url.raw_password:
1029 self.headers[hdrs.AUTHORIZATION] = encode_basic_auth(
1030 url.user or "", url.password or ""
1031 )
1033 def _reset_writer(self, _: object = None) -> None:
1034 self._writer_task = None
1036 def _get_content_length(self) -> int | None:
1037 """Extract and validate Content-Length header value.
1039 Returns parsed Content-Length value or None if not set.
1040 Raises ValueError if header exists but cannot be parsed as an integer.
1041 """
1042 if hdrs.CONTENT_LENGTH not in self.headers:
1043 return None
1045 content_length_hdr = self.headers[hdrs.CONTENT_LENGTH]
1046 if not _DIGITS_RE.fullmatch(content_length_hdr):
1047 raise ValueError(f"Invalid Content-Length header: {content_length_hdr!r}")
1048 return int(content_length_hdr)
1050 @property
1051 def _writer(self) -> asyncio.Task[None] | None:
1052 return self._writer_task
1054 @_writer.setter
1055 def _writer(self, writer: asyncio.Task[None]) -> None:
1056 if self._writer_task is not None:
1057 self._writer_task.remove_done_callback(self._reset_writer)
1058 self._writer_task = writer
1059 writer.add_done_callback(self._reset_writer)
1061 def is_ssl(self) -> bool:
1062 return self.url.scheme in _SSL_SCHEMES
1064 @property
1065 def ssl(self) -> "SSLContext | bool | Fingerprint":
1066 return self._ssl
1068 @property
1069 def connection_key(self) -> ConnectionKey:
1070 url = self.url
1071 return tuple.__new__(
1072 ConnectionKey,
1073 (
1074 url.raw_host or "",
1075 url.port,
1076 url.scheme in _SSL_SCHEMES,
1077 self._ssl,
1078 None,
1079 None,
1080 self.server_hostname,
1081 ),
1082 )
1084 def _update_headers(self, headers: CIMultiDict[str]) -> None:
1085 """Update request headers."""
1086 self.headers: CIMultiDict[str] = CIMultiDict()
1088 # Build the host header
1089 host = self.url.host_port_subcomponent
1091 # host_port_subcomponent is None when the URL is a relative URL.
1092 # but we know we do not have a relative URL here.
1093 assert host is not None
1094 self.headers[hdrs.HOST] = headers.popall(hdrs.HOST, (host,))[0]
1095 self.headers.extend(headers)
1097 def _create_response(
1098 self,
1099 task: asyncio.Task[None] | None,
1100 stream_writer: AbstractStreamWriter,
1101 ) -> ClientResponse:
1102 return self.response_class(
1103 self.method,
1104 self.original_url,
1105 writer=task,
1106 continue100=None,
1107 timer=TimerNoop(),
1108 traces=(),
1109 loop=self.loop,
1110 session=None,
1111 request_headers=self.headers,
1112 original_url=self.original_url,
1113 stream_writer=stream_writer,
1114 )
1116 def _create_writer(self, protocol: BaseProtocol) -> StreamWriter:
1117 return StreamWriter(protocol, self.loop)
1119 def _should_write(self, protocol: BaseProtocol) -> bool:
1120 return protocol.writing_paused
1122 async def _send(self, conn: "Connection") -> ClientResponse:
1123 # Specify request target:
1124 # - CONNECT request must send authority form URI
1125 # - not CONNECT proxy must send absolute form URI
1126 # - most common is origin form URI
1127 if self.method == hdrs.METH_CONNECT:
1128 connect_host = self.url.host_subcomponent
1129 assert connect_host is not None
1130 path = f"{connect_host}:{self.url.port}"
1131 elif self.proxy and not self.is_ssl():
1132 path = str(self.url)
1133 else:
1134 path = self.url.raw_path_qs
1136 protocol = conn.protocol
1137 assert protocol is not None
1138 writer = self._create_writer(protocol)
1140 # set default content-type
1141 if (
1142 self.method in self.POST_METHODS
1143 and (
1144 self._skip_auto_headers is None
1145 or hdrs.CONTENT_TYPE not in self._skip_auto_headers
1146 )
1147 and hdrs.CONTENT_TYPE not in self.headers
1148 ):
1149 self.headers[hdrs.CONTENT_TYPE] = "application/octet-stream"
1151 v = self.version
1152 if hdrs.CONNECTION not in self.headers:
1153 if conn._connector.force_close:
1154 if v == HttpVersion11:
1155 self.headers[hdrs.CONNECTION] = "close"
1156 elif v == HttpVersion10:
1157 self.headers[hdrs.CONNECTION] = "keep-alive"
1159 # status + headers
1160 status_line = f"{self.method} {path} HTTP/{v.major}.{v.minor}"
1162 # Buffer headers for potential coalescing with body
1163 await writer.write_headers(status_line, self.headers)
1165 task: asyncio.Task[None] | None
1166 if self._should_write(protocol):
1167 coro = self._write_bytes(writer, conn, self._get_content_length())
1168 if sys.version_info >= (3, 14):
1169 # Try to write bytes immediately to avoid having to schedule
1170 # the task on the event loop.
1171 loop = asyncio.get_running_loop()
1172 if isinstance(loop, BaseEventLoop):
1173 task = asyncio.create_task(coro, eager_start=True)
1174 else:
1175 task = asyncio.Task(coro, loop=loop, eager_start=True)
1176 elif sys.version_info >= (3, 12):
1177 task = asyncio.Task(
1178 coro, loop=asyncio.get_running_loop(), eager_start=True
1179 )
1180 else:
1181 task = asyncio.create_task(coro)
1182 if task.done():
1183 task = None
1184 else:
1185 self._writer = task
1186 else:
1187 # We have nothing to write because
1188 # - there is no body
1189 # - the protocol does not have writing paused
1190 # - we are not waiting for a 100-continue response
1191 protocol.start_timeout()
1192 writer.set_eof()
1193 task = None
1194 if (tracker := self._upload_tracker) is not None:
1195 # The request went out without a body-writer task: the
1196 # dispatched attempt is trivially complete with zero bytes.
1197 tracker._attempt_finished(self._upload_gen)
1198 self._response = self._create_response(task, stream_writer=writer)
1199 return self._response
1201 async def _write_bytes(
1202 self,
1203 writer: AbstractStreamWriter,
1204 conn: "Connection",
1205 content_length: int | None,
1206 ) -> None:
1207 # Base class never has a body, this will never be run.
1208 assert False
1211class ClientRequestArgs(TypedDict, total=False):
1212 params: Query
1213 headers: CIMultiDict[str]
1214 skip_auto_headers: Iterable[str] | None
1215 data: Any
1216 cookies: BaseCookie[str]
1217 version: HttpVersion
1218 compress: Literal["deflate", "gzip"] | bool
1219 chunked: bool | None
1220 expect100: bool
1221 loop: asyncio.AbstractEventLoop
1222 response_class: type[ClientResponse]
1223 proxy: URL | None
1224 response_params: ResponseParams
1225 timer: BaseTimerContext
1226 timeout: ClientTimeout
1227 session: "ClientSession"
1228 ssl: SSLContext | bool | Fingerprint
1229 proxy_headers: CIMultiDict[str] | None
1230 traces: list["Trace"]
1231 trust_env: bool
1232 server_hostname: str | None
1235class ClientRequest(ClientRequestBase):
1236 _EMPTY_BODY = payload.PAYLOAD_REGISTRY.get(b"", disposition=None)
1237 _body = _EMPTY_BODY
1238 _continue = None # waiter future for '100 Continue' response
1239 _response_params: ResponseParams = None # type: ignore[assignment]
1240 _session: "ClientSession" = None # type: ignore[assignment]
1241 _timeout = ClientTimeout()
1242 _traces: list["Trace"] = () # type: ignore[assignment]
1244 GET_METHODS = {
1245 hdrs.METH_GET,
1246 hdrs.METH_HEAD,
1247 hdrs.METH_OPTIONS,
1248 hdrs.METH_TRACE,
1249 }
1250 DEFAULT_HEADERS = {
1251 hdrs.ACCEPT: "*/*",
1252 hdrs.ACCEPT_ENCODING: _gen_default_accept_encoding(),
1253 }
1255 def __init__(
1256 self,
1257 method: str,
1258 url: URL,
1259 *,
1260 params: Query,
1261 headers: CIMultiDict[str],
1262 skip_auto_headers: Iterable[str] | None,
1263 data: Any,
1264 cookies: BaseCookie[str],
1265 version: HttpVersion,
1266 compress: Literal["deflate", "gzip"] | bool,
1267 chunked: bool | None,
1268 expect100: bool,
1269 loop: asyncio.AbstractEventLoop,
1270 response_class: type[ClientResponse],
1271 proxy: URL | None,
1272 response_params: ResponseParams,
1273 timer: BaseTimerContext,
1274 timeout: ClientTimeout,
1275 session: "ClientSession",
1276 ssl: SSLContext | bool | Fingerprint,
1277 proxy_headers: CIMultiDict[str] | None,
1278 traces: list["Trace"],
1279 trust_env: bool,
1280 server_hostname: str | None,
1281 **kwargs: object,
1282 ):
1283 # kwargs exists so authors of subclasses should expect to pass through unknown
1284 # arguments. This allows us to safely add new arguments in future releases.
1285 # But, we should never receive unknown arguments here in the parent class, this
1286 # would indicate an argument has been named wrong or similar in the subclass.
1287 assert not kwargs, "Unexpected arguments to ClientRequest"
1289 if params:
1290 url = url.extend_query(params)
1291 super().__init__(method, url, headers=headers, loop=loop, ssl=ssl)
1293 if proxy is not None:
1294 assert type(proxy) is URL, proxy
1295 self._session = session
1296 self.chunked = chunked
1297 self.response_class = response_class
1298 self._response_params = response_params
1299 self._timer = timer
1300 self._timeout = timeout
1301 self.server_hostname = server_hostname
1302 self.version = version
1304 self._update_auto_headers(skip_auto_headers)
1305 self._update_cookies(cookies)
1306 self._update_content_encoding(data, compress)
1307 self._update_proxy(proxy, proxy_headers)
1309 self._update_body_from_data(data)
1310 if data is not None or self.method not in self.GET_METHODS:
1311 self._update_transfer_encoding()
1312 self._update_expect_continue(expect100)
1313 self._traces = traces
1315 @property
1316 def body(self) -> payload.Payload:
1317 return self._body
1319 @property
1320 def skip_auto_headers(self) -> CIMultiDict[None]:
1321 return self._skip_auto_headers or CIMultiDict()
1323 @property
1324 def timeout(self) -> ClientTimeout:
1325 """The timeout configuration this request runs under (read-only)."""
1326 return self._timeout
1328 @property
1329 def connection_key(self) -> ConnectionKey:
1330 if proxy_headers := self.proxy_headers:
1331 h: int | None = hash(tuple(proxy_headers.items()))
1332 else:
1333 h = None
1334 url = self.url
1335 return tuple.__new__(
1336 ConnectionKey,
1337 (
1338 url.raw_host or "",
1339 url.port,
1340 url.scheme in _SSL_SCHEMES,
1341 self._ssl,
1342 self.proxy,
1343 h,
1344 self.server_hostname,
1345 ),
1346 )
1348 @property
1349 def session(self) -> "ClientSession":
1350 """Return the ClientSession instance.
1352 This property provides access to the ClientSession that initiated
1353 this request, allowing middleware to make additional requests
1354 using the same session.
1355 """
1356 return self._session
1358 def _update_auto_headers(self, skip_auto_headers: Iterable[str] | None) -> None:
1359 if skip_auto_headers is not None:
1360 self._skip_auto_headers = CIMultiDict(
1361 (hdr, None) for hdr in sorted(skip_auto_headers)
1362 )
1363 used_headers = self.headers.copy()
1364 used_headers.extend(self._skip_auto_headers) # type: ignore[arg-type]
1365 else:
1366 # Fast path when there are no headers to skip
1367 # which is the most common case.
1368 used_headers = self.headers
1370 for hdr, val in self.DEFAULT_HEADERS.items():
1371 if hdr not in used_headers:
1372 self.headers[hdr] = val
1374 if hdrs.USER_AGENT not in used_headers:
1375 self.headers[hdrs.USER_AGENT] = SERVER_SOFTWARE
1377 def _update_cookies(self, cookies: BaseCookie[str]) -> None:
1378 """Update request cookies header."""
1379 if not cookies:
1380 return
1382 c = SimpleCookie()
1383 if hdrs.COOKIE in self.headers:
1384 # parse_cookie_header for RFC 6265 compliant Cookie header parsing
1385 c.update(parse_cookie_header(self.headers.get(hdrs.COOKIE, "")))
1386 del self.headers[hdrs.COOKIE]
1388 for name, value in cookies.items():
1389 # Use helper to preserve coded_value exactly as sent by server
1390 c[name] = preserve_morsel_with_coded_value(value)
1392 self.headers[hdrs.COOKIE] = c.output(header="", sep=";").strip()
1394 def _update_content_encoding(
1395 self, data: Any, compress: bool | Literal["deflate", "gzip"]
1396 ) -> None:
1397 """Set request content encoding."""
1398 self.compress = None
1399 if not data:
1400 return
1402 if self.headers.get(hdrs.CONTENT_ENCODING):
1403 if compress:
1404 raise ValueError(
1405 "compress can not be set if Content-Encoding header is set"
1406 )
1407 elif compress:
1408 if isinstance(compress, str) and compress not in {"deflate", "gzip"}:
1409 raise ValueError(
1410 "compress must be one of True, False, 'deflate', or 'gzip'"
1411 )
1412 self.compress = compress if isinstance(compress, str) else "deflate"
1413 self.headers[hdrs.CONTENT_ENCODING] = self.compress
1414 self.chunked = True # enable chunked, no need to deal with length
1416 def _update_transfer_encoding(self) -> None:
1417 """Analyze transfer-encoding header."""
1418 te = self.headers.get(hdrs.TRANSFER_ENCODING, "").lower()
1420 if "chunked" in te:
1421 if self.chunked:
1422 raise ValueError(
1423 "chunked can not be set "
1424 'if "Transfer-Encoding: chunked" header is set'
1425 )
1427 elif self.chunked:
1428 if hdrs.CONTENT_LENGTH in self.headers:
1429 raise ValueError(
1430 "chunked can not be set if Content-Length header is set"
1431 )
1433 self.headers[hdrs.TRANSFER_ENCODING] = "chunked"
1435 def _update_body_from_data(self, body: Any) -> None:
1436 """Update request body from data."""
1437 if body is None:
1438 self._body = self._EMPTY_BODY
1439 # Set Content-Length to 0 when body is None for methods that expect a body
1440 if (
1441 self.method not in self.GET_METHODS
1442 and not self.chunked
1443 and hdrs.CONTENT_LENGTH not in self.headers
1444 ):
1445 self.headers[hdrs.CONTENT_LENGTH] = "0"
1446 return
1448 # FormData
1449 if isinstance(body, FormData):
1450 body = body()
1451 else:
1452 try:
1453 body = payload.PAYLOAD_REGISTRY.get(body, disposition=None)
1454 except payload.LookupError:
1455 boundary = None
1456 if hdrs.CONTENT_TYPE in self.headers:
1457 boundary = parse_mimetype(
1458 self.headers[hdrs.CONTENT_TYPE]
1459 ).parameters.get("boundary")
1460 body = FormData(body, boundary=boundary)()
1462 self._body = body
1464 # enable chunked encoding if needed
1465 if not self.chunked and hdrs.CONTENT_LENGTH not in self.headers:
1466 if (size := body.size) is not None:
1467 self.headers[hdrs.CONTENT_LENGTH] = str(size)
1468 else:
1469 self.chunked = True
1471 # copy payload headers
1472 assert body.headers
1473 headers = self.headers
1474 skip_headers = self._skip_auto_headers
1475 for key, value in body.headers.items():
1476 if key in headers or (skip_headers is not None and key in skip_headers):
1477 continue
1478 headers[key] = value
1480 def _update_body(self, body: Any) -> None:
1481 """Update request body after its already been set."""
1482 # Remove existing Content-Length header since body is changing
1483 if hdrs.CONTENT_LENGTH in self.headers:
1484 del self.headers[hdrs.CONTENT_LENGTH]
1486 # Remove existing Transfer-Encoding header to avoid conflicts
1487 if self.chunked and hdrs.TRANSFER_ENCODING in self.headers:
1488 del self.headers[hdrs.TRANSFER_ENCODING]
1490 # Now update the body using the existing method
1491 self._update_body_from_data(body)
1493 # Update transfer encoding headers if needed (same logic as __init__)
1494 if body is not None or self.method not in self.GET_METHODS:
1495 self._update_transfer_encoding()
1497 async def update_body(self, body: Any) -> None:
1498 """
1499 Update request body and close previous payload if needed.
1501 This method safely updates the request body by first closing any existing
1502 payload to prevent resource leaks, then setting the new body.
1504 IMPORTANT: Always use this method instead of setting request.body directly.
1505 Direct assignment to request.body will leak resources if the previous body
1506 contains file handles, streams, or other resources that need cleanup.
1508 Args:
1509 body: The new body content. Can be:
1510 - bytes/bytearray: Raw binary data
1511 - str: Text data (will be encoded using charset from Content-Type)
1512 - FormData: Form data that will be encoded as multipart/form-data
1513 - Payload: A pre-configured payload object
1514 - AsyncIterable: An async iterable of bytes chunks
1515 - File-like object: Will be read and sent as binary data
1516 - None: Clears the body
1518 Usage:
1519 # CORRECT: Use update_body
1520 await request.update_body(b"new request data")
1522 # WRONG: Don't set body directly
1523 # request.body = b"new request data" # This will leak resources!
1525 # Update with form data
1526 form_data = FormData()
1527 form_data.add_field('field', 'value')
1528 await request.update_body(form_data)
1530 # Clear body
1531 await request.update_body(None)
1533 Note:
1534 This method is async because it may need to close file handles or
1535 other resources associated with the previous payload. Always await
1536 this method to ensure proper cleanup.
1538 Warning:
1539 Setting request.body directly is highly discouraged and can lead to:
1540 - Resource leaks (unclosed file handles, streams)
1541 - Memory leaks (unreleased buffers)
1542 - Unexpected behavior with streaming payloads
1544 It is not recommended to change the payload type in middleware. If the
1545 body was already set (e.g., as bytes), it's best to keep the same type
1546 rather than converting it (e.g., to str) as this may result in unexpected
1547 behavior.
1549 See Also:
1550 - update_body_from_data: Synchronous body update without cleanup
1551 - body property: Direct body access (STRONGLY DISCOURAGED)
1553 """
1554 # Close existing payload if it exists and needs closing
1555 if self._body is not None:
1556 await self._body.close()
1557 self._update_body(body)
1559 def _update_expect_continue(self, expect: bool = False) -> None:
1560 if expect:
1561 self.headers[hdrs.EXPECT] = "100-continue"
1562 elif (
1563 hdrs.EXPECT in self.headers
1564 and self.headers[hdrs.EXPECT].lower() == "100-continue"
1565 ):
1566 expect = True
1568 if expect:
1569 self._continue = self.loop.create_future()
1571 def _update_proxy(
1572 self,
1573 proxy: URL | None,
1574 proxy_headers: CIMultiDict[str] | None,
1575 ) -> None:
1576 if proxy is None:
1577 self.proxy = None
1578 self.proxy_headers = None
1579 return
1581 if proxy.scheme not in HTTP_AND_EMPTY_SCHEMA_SET:
1582 raise ValueError(
1583 f"aiohttp only supports http(s) proxies (got: {proxy.scheme!r}).\n"
1584 "See third-party libraries for other proxy schemes."
1585 )
1587 # URL-embedded credentials on the proxy map to Proxy-Authorization.
1588 if proxy.raw_user or proxy.raw_password:
1589 auth_header = encode_basic_auth(proxy.user or "", proxy.password or "")
1590 if proxy_headers is None:
1591 proxy_headers = CIMultiDict()
1592 proxy_headers.setdefault(hdrs.PROXY_AUTHORIZATION, auth_header)
1593 proxy = proxy.with_user(None)
1594 self.proxy = proxy
1595 self.proxy_headers = proxy_headers
1597 def _create_response(
1598 self,
1599 task: asyncio.Task[None] | None,
1600 stream_writer: AbstractStreamWriter,
1601 ) -> ClientResponse:
1602 return self.response_class(
1603 self.method,
1604 self.original_url,
1605 writer=task,
1606 continue100=self._continue,
1607 timer=self._timer,
1608 traces=self._traces,
1609 loop=self.loop,
1610 session=self._session,
1611 request_headers=self.headers,
1612 original_url=self.original_url,
1613 stream_writer=stream_writer,
1614 )
1616 def _create_writer(self, protocol: BaseProtocol) -> StreamWriter:
1617 writer = StreamWriter(
1618 protocol,
1619 self.loop,
1620 on_chunk_sent=(
1621 functools.partial(self._on_chunk_request_sent, self.method, self.url)
1622 if self._traces
1623 else None
1624 ),
1625 on_headers_sent=(
1626 functools.partial(self._on_headers_request_sent, self.method, self.url)
1627 if self._traces
1628 else None
1629 ),
1630 )
1632 if self.compress:
1633 writer.enable_compression(self.compress)
1635 if self.chunked is not None:
1636 writer.enable_chunking()
1637 return writer
1639 def _should_write(self, protocol: BaseProtocol) -> bool:
1640 return (
1641 self.body.size != 0 or self._continue is not None or protocol.writing_paused
1642 )
1644 async def _write_bytes(
1645 self,
1646 writer: AbstractStreamWriter,
1647 conn: "Connection",
1648 content_length: int | None,
1649 ) -> None:
1650 """
1651 Write the request body to the connection stream.
1653 This method handles writing different types of request bodies:
1654 1. Payload objects (using their specialized write_with_length method)
1655 2. Bytes/bytearray objects
1656 3. Iterable body content
1658 Args:
1659 writer: The stream writer to write the body to
1660 conn: The connection being used for this request
1661 content_length: Optional maximum number of bytes to write from the body
1662 (None means write the entire body)
1664 The method properly handles:
1665 - Waiting for 100-Continue responses if required
1666 - Content length constraints for chunked encoding
1667 - Error handling for network issues, cancellation, and other exceptions
1668 - Signaling EOF and timeout management
1669 - Upload progress reporting to the request's UploadTracker
1671 Raises:
1672 ClientOSError: When there's an OS-level error writing the body
1673 ClientConnectionError: When there's a general connection error
1674 asyncio.CancelledError: When the operation is cancelled
1676 """
1677 tracker = self._upload_tracker
1678 gen = self._upload_gen
1679 if tracker is not None:
1680 writer.on_body_write = functools.partial(tracker._add_bytes, gen)
1681 tracker._attempt_writing(gen, writer)
1682 try:
1683 # 100 response
1684 if self._continue is not None:
1685 # Force headers to be sent before waiting for 100-continue
1686 writer.send_headers()
1687 await writer.drain()
1688 await self._continue
1690 protocol = conn.protocol
1691 assert protocol is not None
1692 try:
1693 await self._body.write_with_length(writer, content_length)
1694 except OSError as underlying_exc:
1695 reraised_exc = underlying_exc
1697 # Distinguish between timeout and other OS errors for better error reporting
1698 exc_is_not_timeout = underlying_exc.errno is not None or not isinstance(
1699 underlying_exc, asyncio.TimeoutError
1700 )
1701 if exc_is_not_timeout:
1702 reraised_exc = ClientOSError(
1703 underlying_exc.errno,
1704 f"Can not write request body for {self.url !s}",
1705 )
1707 set_exception(protocol, reraised_exc, underlying_exc)
1708 if tracker is not None:
1709 tracker._attempt_failed(gen, reraised_exc)
1710 except Exception as underlying_exc:
1711 wrapped_exc = ClientConnectionError(
1712 "Failed to send bytes into the underlying connection "
1713 f"{conn !s}: {underlying_exc!r}",
1714 )
1715 set_exception(protocol, wrapped_exc, underlying_exc)
1716 if tracker is not None:
1717 tracker._attempt_failed(gen, wrapped_exc)
1718 else:
1719 # Successfully wrote the body, signal EOF and start response timeout
1720 await writer.write_eof()
1721 if tracker is not None:
1722 tracker._attempt_finished(gen)
1723 protocol.start_timeout()
1724 except BaseException as underlying_exc:
1725 # Cancellation, or a failure escaping the inner handlers
1726 # (100-continue preamble, write_eof), leaves the body unsent:
1727 # the connection can't be reused, and the tracker must still
1728 # be notified. Record the attempt first: the tracker
1729 # transitions cannot fail, while conn.close() conceivably can.
1730 if tracker is not None:
1731 tracker._attempt_failed(
1732 gen,
1733 (
1734 None
1735 if isinstance(underlying_exc, asyncio.CancelledError)
1736 else underlying_exc
1737 ),
1738 )
1739 conn.close()
1740 raise
1742 async def _close(self) -> None:
1743 if self._writer_task is not None:
1744 try:
1745 await self._writer_task
1746 except asyncio.CancelledError:
1747 if (task := asyncio.current_task()) and task.cancelling():
1748 raise
1750 def _terminate(self) -> None:
1751 if self._writer_task is not None:
1752 if not self.loop.is_closed():
1753 self._writer_task.cancel()
1754 self._writer_task.remove_done_callback(self._reset_writer)
1755 self._writer_task = None
1757 async def _on_chunk_request_sent(self, method: str, url: URL, chunk: bytes) -> None:
1758 for trace in self._traces:
1759 await trace.send_request_chunk_sent(method, url, chunk)
1761 async def _on_headers_request_sent(
1762 self, method: str, url: URL, headers: "CIMultiDict[str]"
1763 ) -> None:
1764 for trace in self._traces:
1765 await trace.send_request_headers(method, url, headers)