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, Mapping
13from hashlib import md5, sha1, sha256
14from http.cookies import Morsel, SimpleCookie
15from types import MappingProxyType, TracebackType
16from typing import TYPE_CHECKING, Any, Literal, NamedTuple, Optional, Union
18import attr
19from multidict import CIMultiDict, CIMultiDictProxy, MultiDict, MultiDictProxy
20from yarl import URL
22from . import hdrs, helpers, http, multipart, payload
23from ._cookie_helpers import (
24 parse_cookie_header,
25 parse_set_cookie_headers,
26 preserve_morsel_with_coded_value,
27)
28from .abc import AbstractStreamWriter
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 BaseTimerContext,
43 BasicAuth,
44 HeadersMixin,
45 TimerNoop,
46 _basic_auth_no_warn,
47 noop,
48 reify,
49 sentinel,
50 set_exception,
51 set_result,
52)
53from .http import (
54 SERVER_SOFTWARE,
55 HttpVersion,
56 HttpVersion10,
57 HttpVersion11,
58 StreamWriter,
59)
60from .streams import StreamReader
61from .typedefs import (
62 DEFAULT_JSON_DECODER,
63 JSONDecoder,
64 LooseCookies,
65 LooseHeaders,
66 Query,
67 RawHeaders,
68)
70if TYPE_CHECKING:
71 import ssl
72 from ssl import SSLContext
73else:
74 try:
75 import ssl
76 from ssl import SSLContext
77 except ImportError: # pragma: no cover
78 ssl = None # type: ignore[assignment]
79 SSLContext = object # type: ignore[misc,assignment]
82__all__ = ("ClientRequest", "ClientResponse", "RequestInfo", "Fingerprint")
85if TYPE_CHECKING:
86 from .client import ClientSession
87 from .connector import Connection
88 from .tracing import Trace
91_CONNECTION_CLOSED_EXCEPTION = ClientConnectionError("Connection closed")
92_CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
93json_re = re.compile(r"^application/(?:[\w.+-]+?\+)?json")
94_DIGITS_RE = re.compile(r"\d+", re.ASCII)
95_LINK_PARAM_RE = re.compile(r"^([^\s=]+)\s*=\s*(?:(['\"])(.*?)\2|(\S*))$", re.M)
98def _gen_default_accept_encoding() -> str:
99 encodings = [
100 "gzip",
101 "deflate",
102 ]
103 if HAS_BROTLI:
104 encodings.append("br")
105 if HAS_ZSTD:
106 encodings.append("zstd")
107 return ", ".join(encodings)
110@attr.s(auto_attribs=True, frozen=True, slots=True)
111class ContentDisposition:
112 type: str | None
113 parameters: "MappingProxyType[str, str]"
114 filename: str | None
117class _RequestInfo(NamedTuple):
118 url: URL
119 method: str
120 headers: "CIMultiDictProxy[str]"
121 real_url: URL
124class RequestInfo(_RequestInfo):
126 def __new__(
127 cls,
128 url: URL,
129 method: str,
130 headers: "CIMultiDictProxy[str]",
131 real_url: URL | _SENTINEL = sentinel,
132 ) -> "RequestInfo":
133 """Create a new RequestInfo instance.
135 For backwards compatibility, the real_url parameter is optional.
136 """
137 return tuple.__new__(
138 cls, (url, method, headers, url if real_url is sentinel else real_url)
139 )
142class Fingerprint:
143 HASHFUNC_BY_DIGESTLEN = {
144 16: md5,
145 20: sha1,
146 32: sha256,
147 }
149 def __init__(self, fingerprint: bytes) -> None:
150 digestlen = len(fingerprint)
151 hashfunc = self.HASHFUNC_BY_DIGESTLEN.get(digestlen)
152 if not hashfunc:
153 raise ValueError("fingerprint has invalid length")
154 elif hashfunc is md5 or hashfunc is sha1:
155 raise ValueError("md5 and sha1 are insecure and not supported. Use sha256.")
156 self._hashfunc = hashfunc
157 self._fingerprint = fingerprint
159 @property
160 def fingerprint(self) -> bytes:
161 return self._fingerprint
163 def check(self, transport: asyncio.Transport) -> None:
164 if not transport.get_extra_info("sslcontext"):
165 return
166 sslobj = transport.get_extra_info("ssl_object")
167 cert = sslobj.getpeercert(binary_form=True)
168 got = self._hashfunc(cert).digest()
169 if got != self._fingerprint:
170 host, port, *_ = transport.get_extra_info("peername")
171 raise ServerFingerprintMismatch(self._fingerprint, got, host, port)
174if ssl is not None:
175 SSL_ALLOWED_TYPES = (ssl.SSLContext, bool, Fingerprint, type(None))
176else: # pragma: no cover
177 SSL_ALLOWED_TYPES = (bool, type(None))
180def _merge_ssl_params(
181 ssl: Union["SSLContext", bool, Fingerprint],
182 verify_ssl: bool | None,
183 ssl_context: Optional["SSLContext"],
184 fingerprint: bytes | None,
185) -> Union["SSLContext", bool, Fingerprint]:
186 if ssl is None:
187 ssl = True # Double check for backwards compatibility
188 if verify_ssl is not None and not verify_ssl:
189 warnings.warn(
190 "verify_ssl is deprecated, use ssl=False instead",
191 DeprecationWarning,
192 stacklevel=3,
193 )
194 if ssl is not True:
195 raise ValueError(
196 "verify_ssl, ssl_context, fingerprint and ssl "
197 "parameters are mutually exclusive"
198 )
199 else:
200 ssl = False
201 if ssl_context is not None:
202 warnings.warn(
203 "ssl_context is deprecated, use ssl=context instead",
204 DeprecationWarning,
205 stacklevel=3,
206 )
207 if ssl is not True:
208 raise ValueError(
209 "verify_ssl, ssl_context, fingerprint and ssl "
210 "parameters are mutually exclusive"
211 )
212 else:
213 ssl = ssl_context
214 if fingerprint is not None:
215 warnings.warn(
216 "fingerprint is deprecated, use ssl=Fingerprint(fingerprint) instead",
217 DeprecationWarning,
218 stacklevel=3,
219 )
220 if ssl is not True:
221 raise ValueError(
222 "verify_ssl, ssl_context, fingerprint and ssl "
223 "parameters are mutually exclusive"
224 )
225 else:
226 ssl = Fingerprint(fingerprint)
227 if not isinstance(ssl, SSL_ALLOWED_TYPES):
228 raise TypeError(
229 "ssl should be SSLContext, bool, Fingerprint or None, "
230 f"got {ssl!r} instead."
231 )
232 return ssl
235_SSL_SCHEMES = frozenset(("https", "wss"))
238# ConnectionKey is a NamedTuple because it is used as a key in a dict
239# and a set in the connector. Since a NamedTuple is a tuple it uses
240# the fast native tuple __hash__ and __eq__ implementation in CPython.
241class ConnectionKey(NamedTuple):
242 # the key should contain an information about used proxy / TLS
243 # to prevent reusing wrong connections from a pool
244 host: str
245 port: int | None
246 is_ssl: bool
247 ssl: SSLContext | bool | Fingerprint
248 proxy: URL | None
249 proxy_auth: BasicAuth | None
250 proxy_headers_hash: int | None # hash(CIMultiDict)
251 server_hostname: str | None = None
254def _is_expected_content_type(
255 response_content_type: str, expected_content_type: str
256) -> bool:
257 if expected_content_type == "application/json":
258 return json_re.match(response_content_type) is not None
259 return expected_content_type in response_content_type
262def _warn_if_unclosed_payload(payload: payload.Payload, stacklevel: int = 2) -> None:
263 """Warn if the payload is not closed.
265 Callers must check that the body is a Payload before calling this method.
267 Args:
268 payload: The payload to check
269 stacklevel: Stack level for the warning (default 2 for direct callers)
270 """
271 if not payload.autoclose and not payload.consumed:
272 warnings.warn(
273 "The previous request body contains unclosed resources. "
274 "Use await request.update_body() instead of setting request.body "
275 "directly to properly close resources and avoid leaks.",
276 ResourceWarning,
277 stacklevel=stacklevel,
278 )
281# Unflushed checkpoints kept before forcing a refresh in _add_bytes.
282_MAX_UPLOAD_CHECKPOINTS = 256
285class UploadTracker:
286 """Tracks upload progress of a single client request.
288 Pass a fresh instance via the request's ``upload_tracker`` argument and
289 read ``bytes_written``, ``attempts`` and ``upload_complete`` while the
290 request runs.
292 Must be created inside a running event loop. A tracker observes exactly
293 one request: passing it to a second request raises :exc:`RuntimeError`.
294 """
296 def __init__(self) -> None:
297 self.attempts = 0
298 self.upload_complete: asyncio.Future[None] = (
299 asyncio.get_running_loop().create_future()
300 )
301 self._bound = False
302 # Outcome of the latest attempt: finished, failed with _exc, or
303 # neither (never sent, or cut short).
304 self._finished = False
305 self._exc: BaseException | None = None
306 # Set once the request reaches a terminal point (returned or raised);
307 # no further attempts can start after that.
308 self._final = False
309 # Set while the current attempt's body is being written.
310 self._writer: AbstractStreamWriter | None = None
311 # Payload bytes handed to the writer for the current attempt.
312 self._accepted = 0
313 # Payload bytes known to have left the transport for the kernel.
314 self._flushed = 0
315 # (payload total, wire total) pairs: the last fully flushed point
316 # first, then accepted chunks not yet out of the transport's buffer.
317 self._checkpoints: deque[tuple[int, int]] = deque()
319 @property
320 def bytes_written(self) -> int:
321 """Payload bytes of the current attempt sent to the kernel.
323 Computed from the transport's unsent buffer on access, so it
324 excludes bytes still queued in the event loop; it cannot see past
325 the kernel's own socket buffering.
326 """
327 self._refresh_flushed()
328 return self._flushed
330 def _refresh_flushed(self) -> None:
331 writer = self._writer
332 if writer is None:
333 return
334 transport = writer.transport
335 if transport is None or transport.is_closing():
336 # Torn down: bytes dropped from the buffer were never sent,
337 # so keep the last confirmed value.
338 return
339 wire_sent = writer.output_size - transport.get_write_buffer_size()
340 checkpoints = self._checkpoints
341 while len(checkpoints) > 1 and checkpoints[1][1] <= wire_sent:
342 checkpoints.popleft()
343 # A live _writer implies the sentinel pair from _attempt_writing
344 # is present, and the loop above never pops the head.
345 flushed, wire = checkpoints[0]
346 if len(checkpoints) > 1 and wire_sent > wire:
347 # Inside a partially flushed chunk: scale the payload bytes
348 # linearly over the chunk's wire bytes (exact for identity
349 # writes, an estimate under compression).
350 payload_cum, wire_cum = checkpoints[1]
351 flushed += (payload_cum - flushed) * (wire_sent - wire) // (wire_cum - wire)
352 self._flushed = flushed
354 def _bind(self) -> None:
355 if self._bound:
356 raise RuntimeError("UploadTracker is already bound to a request")
357 self._bound = True
359 def _attempt_started(self) -> int:
360 """A new upload attempt is dispatched; returns its generation token.
362 Called before connecting, so a resend failing early settles as
363 aborted instead of reporting the superseded attempt's outcome.
364 """
365 self.attempts += 1
366 self._writer = None
367 self._accepted = 0
368 self._flushed = 0
369 self._checkpoints.clear()
370 self._finished = False
371 self._exc = None
372 return self.attempts
374 def _attempt_writing(self, gen: int, writer: AbstractStreamWriter) -> None:
375 """The attempt's body write has begun; settling now defers to it."""
376 if gen == self.attempts:
377 self._writer = writer
378 # Baseline past any bytes already on the wire (e.g. headers
379 # sent for the 100-continue preamble).
380 self._checkpoints.append((0, writer.output_size))
382 def _add_bytes(self, gen: int, size: int, wire_position: int) -> None:
383 # A stale writer of a superseded attempt (e.g. still being torn
384 # down while a redirect resends the body) must not corrupt the counter.
385 if gen == self.attempts and self._writer is not None:
386 self._accepted += size
387 checkpoints = self._checkpoints
388 # A compressor may swallow a chunk without producing output;
389 # the wire position is then unchanged and no checkpoint is due.
390 if wire_position > checkpoints[-1][1]:
391 checkpoints.append((self._accepted, wire_position))
392 # Bound the queue when nobody polls bytes_written: flushed
393 # entries are only dropped on refresh.
394 if len(checkpoints) > _MAX_UPLOAD_CHECKPOINTS:
395 self._refresh_flushed()
397 def _attempt_finished(self, gen: int) -> None:
398 if gen == self.attempts:
399 self._finished = True
400 # The body was written in full; the tail still in transit is
401 # ordered ahead of the response that completes the request.
402 self._flushed = self._accepted
403 self._checkpoints.clear()
404 self._writer = None
405 if self._final:
406 self._settle()
408 def _attempt_failed(self, gen: int, exc: BaseException | None) -> None:
409 """The attempt did not send the body in full.
411 ``exc`` is the upload error, or ``None`` when the attempt was cut
412 short (cancelled) rather than failed.
413 """
414 if gen == self.attempts:
415 self._exc = exc
416 # Keep the last kernel-confirmed value and drop the writer.
417 self._refresh_flushed()
418 self._checkpoints.clear()
419 self._writer = None
420 if self._final:
421 self._settle()
423 def _finalize(self) -> None:
424 """Mark the request terminal: no further attempts will start.
426 Settles upload_complete unless the final attempt is still writing,
427 in which case the attempt's own terminal event settles it.
428 """
429 self._final = True
430 if self._writer is None:
431 self._settle()
433 def _settle(self) -> None:
434 fut = self.upload_complete
435 if fut.done():
436 return
437 if self._finished:
438 fut.set_result(None)
439 return
440 if self._exc is not None:
441 fut.set_exception(self._exc)
442 else:
443 fut.set_exception(UploadAbortedError("The request body was not fully sent"))
444 # Avoid 'exception was never retrieved' when doing a .done() poll.
445 fut.exception()
448class ClientResponse(HeadersMixin):
450 # Some of these attributes are None when created,
451 # but will be set by the start() method.
452 # As the end user will likely never see the None values, we cheat the types below.
453 # from the Status-Line of the response
454 version: HttpVersion | None = None # HTTP-Version
455 status: int = None # type: ignore[assignment] # Status-Code
456 reason: str | None = None # Reason-Phrase
458 content: StreamReader = None # type: ignore[assignment] # Payload stream
459 _body: bytes | None = None
460 _headers: CIMultiDictProxy[str] = None # type: ignore[assignment]
461 _history: tuple["ClientResponse", ...] = ()
462 _raw_headers: RawHeaders = None # type: ignore[assignment]
464 _connection: Optional["Connection"] = None # current connection
465 _cookies: SimpleCookie | None = None
466 _raw_cookie_headers: tuple[str, ...] | None = None
467 _continue: Optional["asyncio.Future[bool]"] = None
468 _source_traceback: traceback.StackSummary | None = None
469 _session: Optional["ClientSession"] = None
470 # set up by ClientRequest after ClientResponse object creation
471 # post-init stage allows to not change ctor signature
472 _closed = True # to allow __del__ for non-initialized properly response
473 _released = False
474 _in_context = False
476 _resolve_charset: Callable[["ClientResponse", bytes], str] = lambda *_: "utf-8"
478 __writer: Optional["asyncio.Task[None]"] = None
479 _stream_writer: Optional[AbstractStreamWriter] = None
480 _output_size: int = 0
481 _upload_complete: Optional[asyncio.Future[None]] = None
483 def __init__(
484 self,
485 method: str,
486 url: URL,
487 *,
488 writer: "asyncio.Task[None] | None",
489 continue100: Optional["asyncio.Future[bool]"],
490 timer: BaseTimerContext,
491 request_info: RequestInfo,
492 traces: list["Trace"],
493 loop: asyncio.AbstractEventLoop,
494 session: "ClientSession",
495 stream_writer: AbstractStreamWriter,
496 ) -> None:
497 # URL forbids subclasses, so a simple type check is enough.
498 assert type(url) is URL
500 self.method = method
502 self._real_url = url
503 self._url = url.with_fragment(None) if url.raw_fragment else url
504 if writer is None: # Request already sent
505 self._output_size = stream_writer.output_size
506 else:
507 self._stream_writer = stream_writer
508 self._writer = writer
509 if continue100 is not None:
510 self._continue = continue100
511 self._request_info = request_info
512 self._timer = timer if timer is not None else TimerNoop()
513 self._cache: dict[str, Any] = {}
514 self._traces = traces
515 self._loop = loop
516 # Save reference to _resolve_charset, so that get_encoding() will still
517 # work after the response has finished reading the body.
518 # TODO: Fix session=None in tests (see ClientRequest.__init__).
519 if session is not None:
520 # store a reference to session #1985
521 self._session = session
522 self._resolve_charset = session._resolve_charset
523 if loop.get_debug():
524 self._source_traceback = traceback.extract_stack(sys._getframe(1))
526 def __reset_writer(self, _: object = None) -> None:
527 self.__writer = None
528 if self._stream_writer is not None:
529 self._output_size = self._stream_writer.output_size
530 self._stream_writer = None
531 if self._upload_complete is not None and not self._upload_complete.done():
532 self._upload_complete.set_result(None)
534 @property
535 def _writer(self) -> Optional["asyncio.Task[None]"]:
536 """The writer task for streaming data.
538 _writer is only provided for backwards compatibility
539 for subclasses that may need to access it.
540 """
541 return self.__writer
543 @_writer.setter
544 def _writer(self, writer: Optional["asyncio.Task[None]"]) -> None:
545 """Set the writer task for streaming data."""
546 if self.__writer is not None:
547 self.__writer.remove_done_callback(self.__reset_writer)
548 self.__writer = writer
549 if writer is None:
550 return
551 if writer.done():
552 # The writer is already done, so we can clear it immediately.
553 self.__reset_writer()
554 else:
555 writer.add_done_callback(self.__reset_writer)
557 @property
558 def output_size(self) -> int:
559 """Number of bytes sent for this request.
561 .. deprecated:: 3.14.4
562 Use :class:`UploadTracker` instead.
563 """
564 warnings.warn(
565 "ClientResponse.output_size is deprecated, "
566 "use aiohttp.UploadTracker instead",
567 DeprecationWarning,
568 stacklevel=2,
569 )
570 if self._stream_writer is not None:
571 return self._stream_writer.output_size
572 return self._output_size
574 @property
575 def upload_complete(self) -> "asyncio.Future[None]":
576 """Future set when the request body has been fully sent.
578 .. deprecated:: 3.14.4
579 Use :class:`UploadTracker` instead.
580 """
581 warnings.warn(
582 "ClientResponse.upload_complete is deprecated, "
583 "use aiohttp.UploadTracker instead",
584 DeprecationWarning,
585 stacklevel=2,
586 )
587 if self._upload_complete is None:
588 self._upload_complete = self._loop.create_future()
589 if self._stream_writer is None: # upload already finished
590 self._upload_complete.set_result(None)
591 return self._upload_complete
593 @property
594 def cookies(self) -> SimpleCookie:
595 if self._cookies is None:
596 if self._raw_cookie_headers is not None:
597 # Parse cookies for response.cookies (SimpleCookie for backward compatibility)
598 cookies = SimpleCookie()
599 # Use parse_set_cookie_headers for more lenient parsing that handles
600 # malformed cookies better than SimpleCookie.load
601 cookies.update(parse_set_cookie_headers(self._raw_cookie_headers))
602 self._cookies = cookies
603 else:
604 self._cookies = SimpleCookie()
605 return self._cookies
607 @cookies.setter
608 def cookies(self, cookies: SimpleCookie) -> None:
609 self._cookies = cookies
610 # Generate raw cookie headers from the SimpleCookie
611 if cookies:
612 self._raw_cookie_headers = tuple(
613 morsel.OutputString() for morsel in cookies.values()
614 )
615 else:
616 self._raw_cookie_headers = None
618 @reify
619 def url(self) -> URL:
620 return self._url
622 @reify
623 def url_obj(self) -> URL:
624 warnings.warn("Deprecated, use .url #1654", DeprecationWarning, stacklevel=2)
625 return self._url
627 @reify
628 def real_url(self) -> URL:
629 return self._real_url
631 @reify
632 def host(self) -> str:
633 assert self._url.host is not None
634 return self._url.host
636 @reify
637 def headers(self) -> "CIMultiDictProxy[str]":
638 return self._headers
640 @reify
641 def raw_headers(self) -> RawHeaders:
642 return self._raw_headers
644 @reify
645 def request_info(self) -> RequestInfo:
646 return self._request_info
648 @reify
649 def content_disposition(self) -> ContentDisposition | None:
650 raw = self._headers.get(hdrs.CONTENT_DISPOSITION)
651 if raw is None:
652 return None
653 disposition_type, params_dct = multipart.parse_content_disposition(raw)
654 params = MappingProxyType(params_dct)
655 filename = multipart.content_disposition_filename(params)
656 return ContentDisposition(disposition_type, params, filename)
658 def __del__(self, _warnings: Any = warnings) -> None:
659 if self._closed:
660 return
662 if self._connection is not None:
663 self._connection.release()
664 self._cleanup_writer()
666 if self._loop.get_debug():
667 kwargs = {"source": self}
668 _warnings.warn(f"Unclosed response {self!r}", ResourceWarning, **kwargs)
669 context = {"client_response": self, "message": "Unclosed response"}
670 if self._source_traceback:
671 context["source_traceback"] = self._source_traceback
672 self._loop.call_exception_handler(context)
674 def __repr__(self) -> str:
675 out = io.StringIO()
676 ascii_encodable_url = str(self.url)
677 if self.reason:
678 ascii_encodable_reason = self.reason.encode(
679 "ascii", "backslashreplace"
680 ).decode("ascii")
681 else:
682 ascii_encodable_reason = "None"
683 print(
684 f"<ClientResponse({ascii_encodable_url}) [{self.status} {ascii_encodable_reason}]>",
685 file=out,
686 )
687 print(self.headers, file=out)
688 return out.getvalue()
690 @property
691 def connection(self) -> Optional["Connection"]:
692 return self._connection
694 @reify
695 def history(self) -> tuple["ClientResponse", ...]:
696 """A sequence of of responses, if redirects occurred."""
697 return self._history
699 @reify
700 def links(self) -> "MultiDictProxy[MultiDictProxy[str | URL]]":
701 links_str = ", ".join(self.headers.getall("link", []))
703 if not links_str:
704 return MultiDictProxy(MultiDict())
706 links: MultiDict[MultiDictProxy[str | URL]] = MultiDict()
708 for val in re.split(r",(?=\s*<)", links_str):
709 match = re.match(r"\s*<(.*)>(.*)", val)
710 if match is None: # pragma: no cover
711 # the check exists to suppress mypy error
712 continue
713 url, params_str = match.groups()
714 params = params_str.split(";")[1:]
716 link: MultiDict[str | URL] = MultiDict()
718 for param in params:
719 match = _LINK_PARAM_RE.match(param.strip())
720 if match is None: # Malformed param
721 continue
722 key, _, value_quoted, value_unquoted = match.groups()
724 link.add(key, value_unquoted if value_quoted is None else value_quoted)
726 key = link.get("rel", url)
728 link.add("url", self.url.join(URL(url)))
730 links.add(str(key), MultiDictProxy(link))
732 return MultiDictProxy(links)
734 async def start(self, connection: "Connection") -> "ClientResponse":
735 """Start response processing."""
736 self._closed = False
737 self._protocol = connection.protocol
738 self._connection = connection
740 with self._timer:
741 while True:
742 # read response
743 try:
744 protocol = self._protocol
745 message, payload = await protocol.read() # type: ignore[union-attr]
746 except http.HttpProcessingError as exc:
747 raise ClientResponseError(
748 self.request_info,
749 self.history,
750 status=exc.code,
751 message=exc.message,
752 headers=exc.headers,
753 ) from exc
755 if message.code < 100 or message.code > 199 or message.code == 101:
756 break
758 if self._continue is not None:
759 set_result(self._continue, True)
760 self._continue = None
762 # payload eof handler
763 payload.on_eof(self._response_eof)
765 # response status
766 self.version = message.version
767 self.status = message.code
768 self.reason = message.reason
770 # headers
771 self._headers = message.headers # type is CIMultiDictProxy
772 self._raw_headers = message.raw_headers # type is Tuple[bytes, bytes]
774 # payload
775 self.content = payload
777 # cookies
778 if cookie_hdrs := self.headers.getall(hdrs.SET_COOKIE, ()):
779 # Store raw cookie headers for CookieJar
780 self._raw_cookie_headers = tuple(cookie_hdrs)
781 return self
783 def _response_eof(self) -> None:
784 if self._closed:
785 return
787 # protocol could be None because connection could be detached
788 protocol = self._connection and self._connection.protocol
789 if protocol is not None and protocol.upgraded:
790 return
792 self._closed = True
793 self._cleanup_writer()
794 self._release_connection()
796 @property
797 def closed(self) -> bool:
798 return self._closed
800 def close(self) -> None:
801 if not self._released:
802 self._notify_content()
804 self._closed = True
805 if self._loop is None or self._loop.is_closed():
806 return
808 self._cleanup_writer()
809 if self._connection is not None:
810 self._connection.close()
811 self._connection = None
813 def release(self) -> Any:
814 if not self._released:
815 self._notify_content()
817 self._closed = True
819 self._cleanup_writer()
820 self._release_connection()
821 return noop()
823 @property
824 def ok(self) -> bool:
825 """Returns ``True`` if ``status`` is less than ``400``, ``False`` if not.
827 This is **not** a check for ``200 OK`` but a check that the response
828 status is under 400.
829 """
830 return 400 > self.status
832 def raise_for_status(self) -> None:
833 if not self.ok:
834 # reason should always be not None for a started response
835 assert self.reason is not None
837 # If we're in a context we can rely on __aexit__() to release as the
838 # exception propagates.
839 if not self._in_context:
840 self.release()
842 raise ClientResponseError(
843 self.request_info,
844 self.history,
845 status=self.status,
846 message=self.reason,
847 headers=self.headers,
848 )
850 def _release_connection(self) -> None:
851 if self._connection is not None:
852 if self.__writer is None:
853 self._connection.release()
854 self._connection = None
855 else:
856 self.__writer.add_done_callback(lambda f: self._release_connection())
858 async def _wait_released(self) -> None:
859 if self.__writer is not None:
860 try:
861 await self.__writer
862 except asyncio.CancelledError:
863 if (
864 sys.version_info >= (3, 11)
865 and (task := asyncio.current_task())
866 and task.cancelling()
867 ):
868 raise
869 self._release_connection()
871 def _cleanup_writer(self) -> None:
872 if self.__writer is not None:
873 self.__writer.cancel()
874 if self._stream_writer is not None:
875 self._output_size = self._stream_writer.output_size
876 self._stream_writer = None
877 self._session = None
879 def _notify_content(self) -> None:
880 content = self.content
881 if content and content.exception() is None:
882 set_exception(content, _CONNECTION_CLOSED_EXCEPTION)
883 self._released = True
885 async def wait_for_close(self) -> None:
886 if self.__writer is not None:
887 try:
888 await self.__writer
889 except asyncio.CancelledError:
890 if (
891 sys.version_info >= (3, 11)
892 and (task := asyncio.current_task())
893 and task.cancelling()
894 ):
895 raise
896 self.release()
898 async def read(self) -> bytes:
899 """Read response payload."""
900 if self._body is None:
901 try:
902 self._body = await self.content.read()
903 for trace in self._traces:
904 await trace.send_response_chunk_received(
905 self.method, self.url, self._body
906 )
907 except BaseException:
908 self.close()
909 raise
910 elif self._released: # Response explicitly released
911 raise ClientConnectionError("Connection closed")
913 protocol = self._connection and self._connection.protocol
914 if protocol is None or not protocol.upgraded:
915 await self._wait_released() # Underlying connection released
916 return self._body
918 def get_encoding(self) -> str:
919 ctype = self.headers.get(hdrs.CONTENT_TYPE, "").lower()
920 mimetype = helpers.parse_mimetype(ctype)
922 encoding = mimetype.parameters.get("charset")
923 if encoding:
924 with contextlib.suppress(LookupError, ValueError):
925 return codecs.lookup(encoding).name
927 if mimetype.type == "application" and (
928 mimetype.subtype == "json" or mimetype.subtype == "rdap"
929 ):
930 # RFC 7159 states that the default encoding is UTF-8.
931 # RFC 7483 defines application/rdap+json
932 return "utf-8"
934 if self._body is None:
935 raise RuntimeError(
936 "Cannot compute fallback encoding of a not yet read body"
937 )
939 return self._resolve_charset(self, self._body)
941 async def text(self, encoding: str | None = None, errors: str = "strict") -> str:
942 """Read response payload and decode."""
943 if self._body is None:
944 await self.read()
946 if encoding is None:
947 encoding = self.get_encoding()
949 return self._body.decode(encoding, errors=errors) # type: ignore[union-attr]
951 async def json(
952 self,
953 *,
954 encoding: str | None = None,
955 loads: JSONDecoder = DEFAULT_JSON_DECODER,
956 content_type: str | None = "application/json",
957 ) -> Any:
958 """Read and decodes JSON response."""
959 if self._body is None:
960 await self.read()
962 if content_type:
963 ctype = self.headers.get(hdrs.CONTENT_TYPE, "").lower()
964 if not _is_expected_content_type(ctype, content_type):
965 raise ContentTypeError(
966 self.request_info,
967 self.history,
968 status=self.status,
969 message=(
970 "Attempt to decode JSON with unexpected mimetype: %s" % ctype
971 ),
972 headers=self.headers,
973 )
975 stripped = self._body.strip() # type: ignore[union-attr]
976 if not stripped:
977 return None
979 if encoding is None:
980 encoding = self.get_encoding()
982 return loads(stripped.decode(encoding))
984 async def __aenter__(self) -> "ClientResponse":
985 self._in_context = True
986 return self
988 async def __aexit__(
989 self,
990 exc_type: type[BaseException] | None,
991 exc_val: BaseException | None,
992 exc_tb: TracebackType | None,
993 ) -> None:
994 self._in_context = False
995 # similar to _RequestContextManager, we do not need to check
996 # for exceptions, response object can close connection
997 # if state is broken
998 self.release()
999 await self.wait_for_close()
1002class ClientRequest:
1003 GET_METHODS = {
1004 hdrs.METH_GET,
1005 hdrs.METH_HEAD,
1006 hdrs.METH_OPTIONS,
1007 hdrs.METH_TRACE,
1008 }
1009 POST_METHODS = {hdrs.METH_PATCH, hdrs.METH_POST, hdrs.METH_PUT}
1010 ALL_METHODS = GET_METHODS.union(POST_METHODS).union({hdrs.METH_DELETE})
1012 DEFAULT_HEADERS = {
1013 hdrs.ACCEPT: "*/*",
1014 hdrs.ACCEPT_ENCODING: _gen_default_accept_encoding(),
1015 }
1017 # Type of body depends on PAYLOAD_REGISTRY, which is dynamic.
1018 _body: None | payload.Payload = None
1019 auth = None
1020 response = None
1022 __writer: Optional["asyncio.Task[None]"] = None # async task for streaming data
1024 # These class defaults help create_autospec() work correctly.
1025 # If autospec is improved in future, maybe these can be removed.
1026 url = URL()
1027 method = "GET"
1029 _continue = None # waiter future for '100 Continue' response
1031 _upload_tracker: UploadTracker | None = None
1032 _upload_gen = 0
1034 _skip_auto_headers: "CIMultiDict[None] | None" = None
1036 # N.B.
1037 # Adding __del__ method with self._writer closing doesn't make sense
1038 # because _writer is instance method, thus it keeps a reference to self.
1039 # Until writer has finished finalizer will not be called.
1041 def __init__(
1042 self,
1043 method: str,
1044 url: URL,
1045 *,
1046 params: Query = None,
1047 headers: LooseHeaders | None = None,
1048 skip_auto_headers: Iterable[str] | None = None,
1049 data: Any = None,
1050 cookies: LooseCookies | None = None,
1051 auth: BasicAuth | None = None,
1052 version: http.HttpVersion = http.HttpVersion11,
1053 compress: str | bool | None = None,
1054 chunked: bool | None = None,
1055 expect100: bool = False,
1056 loop: asyncio.AbstractEventLoop | None = None,
1057 response_class: type["ClientResponse"] | None = None,
1058 proxy: URL | None = None,
1059 proxy_auth: BasicAuth | None = None,
1060 timer: BaseTimerContext | None = None,
1061 session: Optional["ClientSession"] = None,
1062 ssl: SSLContext | bool | Fingerprint = True,
1063 proxy_headers: LooseHeaders | None = None,
1064 traces: list["Trace"] | None = None,
1065 trust_env: bool = False,
1066 server_hostname: str | None = None,
1067 ):
1068 if loop is None:
1069 loop = asyncio.get_event_loop()
1070 if match := _CONTAINS_CONTROL_CHAR_RE.search(method):
1071 raise ValueError(
1072 f"Method cannot contain non-token characters {method!r} "
1073 f"(found at least {match.group()!r})"
1074 )
1075 # URL forbids subclasses, so a simple type check is enough.
1076 assert type(url) is URL, url
1077 if proxy is not None:
1078 assert type(proxy) is URL, proxy
1079 # FIXME: session is None in tests only, need to fix tests
1080 # assert session is not None
1081 if TYPE_CHECKING:
1082 assert session is not None
1083 self._session = session
1084 if params:
1085 url = url.extend_query(params)
1086 self.original_url = url
1087 self.url = url.with_fragment(None) if url.raw_fragment else url
1088 self.method = method.upper()
1089 self.chunked = chunked
1090 self.compress = compress
1091 self.loop = loop
1092 self.length = None
1093 if response_class is None:
1094 real_response_class = ClientResponse
1095 else:
1096 real_response_class = response_class
1097 self.response_class: type[ClientResponse] = real_response_class
1098 self._timer = timer if timer is not None else TimerNoop()
1099 self._ssl = ssl if ssl is not None else True
1100 self.server_hostname = server_hostname
1102 if loop.get_debug():
1103 self._source_traceback = traceback.extract_stack(sys._getframe(1))
1105 self.update_version(version)
1106 self.update_host(url)
1107 self.update_headers(headers)
1108 self.update_auto_headers(skip_auto_headers)
1109 self.update_cookies(cookies)
1110 self.update_content_encoding(data)
1111 self.update_auth(auth, trust_env)
1112 self.update_proxy(proxy, proxy_auth, proxy_headers)
1114 self.update_body_from_data(data)
1115 if data is not None or self.method not in self.GET_METHODS:
1116 self.update_transfer_encoding()
1117 self.update_expect_continue(expect100)
1118 self._traces = [] if traces is None else traces
1120 def __reset_writer(self, _: object = None) -> None:
1121 self.__writer = None
1123 def _get_content_length(self) -> int | None:
1124 """Extract and validate Content-Length header value.
1126 Returns parsed Content-Length value or None if not set.
1127 Raises ValueError if header exists but cannot be parsed as an integer.
1128 """
1129 if hdrs.CONTENT_LENGTH not in self.headers:
1130 return None
1132 content_length_hdr = self.headers[hdrs.CONTENT_LENGTH]
1133 if not _DIGITS_RE.fullmatch(content_length_hdr):
1134 raise ValueError(f"Invalid Content-Length header: {content_length_hdr!r}")
1135 return int(content_length_hdr)
1137 @property
1138 def skip_auto_headers(self) -> CIMultiDict[None]:
1139 return self._skip_auto_headers or CIMultiDict()
1141 @property
1142 def _writer(self) -> Optional["asyncio.Task[None]"]:
1143 return self.__writer
1145 @_writer.setter
1146 def _writer(self, writer: "asyncio.Task[None]") -> None:
1147 if self.__writer is not None:
1148 self.__writer.remove_done_callback(self.__reset_writer)
1149 self.__writer = writer
1150 writer.add_done_callback(self.__reset_writer)
1152 def is_ssl(self) -> bool:
1153 return self.url.scheme in _SSL_SCHEMES
1155 @property
1156 def ssl(self) -> Union["SSLContext", bool, Fingerprint]:
1157 return self._ssl
1159 @property
1160 def connection_key(self) -> ConnectionKey:
1161 if proxy_headers := self.proxy_headers:
1162 h: int | None = hash(tuple(proxy_headers.items()))
1163 else:
1164 h = None
1165 url = self.url
1166 return tuple.__new__(
1167 ConnectionKey,
1168 (
1169 url.raw_host or "",
1170 url.port,
1171 url.scheme in _SSL_SCHEMES,
1172 self._ssl,
1173 self.proxy,
1174 self.proxy_auth,
1175 h,
1176 self.server_hostname,
1177 ),
1178 )
1180 @property
1181 def host(self) -> str:
1182 ret = self.url.raw_host
1183 assert ret is not None
1184 return ret
1186 @property
1187 def port(self) -> int | None:
1188 return self.url.port
1190 @property
1191 def body(self) -> payload.Payload | Literal[b""]:
1192 """Request body."""
1193 # empty body is represented as bytes for backwards compatibility
1194 return self._body or b""
1196 @body.setter
1197 def body(self, value: Any) -> None:
1198 """Set request body with warning for non-autoclose payloads.
1200 WARNING: This setter must be called from within an event loop and is not
1201 thread-safe. Setting body outside of an event loop may raise RuntimeError
1202 when closing file-based payloads.
1204 DEPRECATED: Direct assignment to body is deprecated and will be removed
1205 in a future version. Use await update_body() instead for proper resource
1206 management.
1207 """
1208 # Close existing payload if present
1209 if self._body is not None:
1210 # Warn if the payload needs manual closing
1211 # stacklevel=3: user code -> body setter -> _warn_if_unclosed_payload
1212 _warn_if_unclosed_payload(self._body, stacklevel=3)
1213 # NOTE: In the future, when we remove sync close support,
1214 # this setter will need to be removed and only the async
1215 # update_body() method will be available. For now, we call
1216 # _close() for backwards compatibility.
1217 self._body._close()
1218 self._update_body(value)
1220 @property
1221 def request_info(self) -> RequestInfo:
1222 headers: CIMultiDictProxy[str] = CIMultiDictProxy(self.headers)
1223 # These are created on every request, so we use a NamedTuple
1224 # for performance reasons. We don't use the RequestInfo.__new__
1225 # method because it has a different signature which is provided
1226 # for backwards compatibility only.
1227 return tuple.__new__(
1228 RequestInfo, (self.url, self.method, headers, self.original_url)
1229 )
1231 @property
1232 def session(self) -> "ClientSession":
1233 """Return the ClientSession instance.
1235 This property provides access to the ClientSession that initiated
1236 this request, allowing middleware to make additional requests
1237 using the same session.
1238 """
1239 return self._session
1241 def update_host(self, url: URL) -> None:
1242 """Update destination host, port and connection type (ssl)."""
1243 # get host/port
1244 if not url.raw_host:
1245 raise InvalidURL(url)
1247 # basic auth info
1248 if url.raw_user or url.raw_password:
1249 self.auth = _basic_auth_no_warn(url.user or "", url.password or "")
1251 def update_version(self, version: http.HttpVersion | str) -> None:
1252 """Convert request version to two elements tuple.
1254 parser HTTP version '1.1' => (1, 1)
1255 """
1256 if isinstance(version, str):
1257 v = [part.strip() for part in version.split(".", 1)]
1258 try:
1259 version = http.HttpVersion(int(v[0]), int(v[1]))
1260 except ValueError:
1261 raise ValueError(
1262 f"Can not parse http version number: {version}"
1263 ) from None
1264 self.version = version
1266 def update_headers(self, headers: LooseHeaders | None) -> None:
1267 """Update request headers."""
1268 self.headers: CIMultiDict[str] = CIMultiDict()
1270 # Build the host header
1271 host = self.url.host_port_subcomponent
1273 # host_port_subcomponent is None when the URL is a relative URL.
1274 # but we know we do not have a relative URL here.
1275 assert host is not None
1276 self.headers[hdrs.HOST] = host
1278 if not headers:
1279 return
1281 if isinstance(headers, (dict, MultiDictProxy, MultiDict)):
1282 headers = headers.items()
1284 for key, value in headers: # type: ignore[str-unpack]
1285 # A special case for Host header
1286 if key in hdrs.HOST_ALL:
1287 self.headers[key] = value
1288 else:
1289 self.headers.add(key, value)
1291 def update_auto_headers(self, skip_auto_headers: Iterable[str] | None) -> None:
1292 if skip_auto_headers is not None:
1293 self._skip_auto_headers = CIMultiDict(
1294 (hdr, None) for hdr in sorted(skip_auto_headers)
1295 )
1296 used_headers = self.headers.copy()
1297 used_headers.extend(self._skip_auto_headers) # type: ignore[arg-type]
1298 else:
1299 # Fast path when there are no headers to skip
1300 # which is the most common case.
1301 used_headers = self.headers
1303 for hdr, val in self.DEFAULT_HEADERS.items():
1304 if hdr not in used_headers:
1305 self.headers[hdr] = val
1307 if hdrs.USER_AGENT not in used_headers:
1308 self.headers[hdrs.USER_AGENT] = SERVER_SOFTWARE
1310 def update_cookies(self, cookies: LooseCookies | None) -> None:
1311 """Update request cookies header."""
1312 if not cookies:
1313 return
1315 c = SimpleCookie()
1316 if hdrs.COOKIE in self.headers:
1317 # parse_cookie_header for RFC 6265 compliant Cookie header parsing
1318 c.update(parse_cookie_header(self.headers.get(hdrs.COOKIE, "")))
1319 del self.headers[hdrs.COOKIE]
1321 if isinstance(cookies, Mapping):
1322 iter_cookies = cookies.items()
1323 else:
1324 iter_cookies = cookies # type: ignore[assignment]
1325 for name, value in iter_cookies:
1326 if isinstance(value, Morsel):
1327 # Use helper to preserve coded_value exactly as sent by server
1328 c[name] = preserve_morsel_with_coded_value(value)
1329 else:
1330 c[name] = value # type: ignore[assignment]
1332 self.headers[hdrs.COOKIE] = c.output(header="", sep=";").strip()
1334 def update_content_encoding(self, data: Any) -> None:
1335 """Set request content encoding."""
1336 if not data:
1337 # Don't compress an empty body.
1338 self.compress = None
1339 return
1341 if self.headers.get(hdrs.CONTENT_ENCODING):
1342 if self.compress:
1343 raise ValueError(
1344 "compress can not be set if Content-Encoding header is set"
1345 )
1346 elif self.compress:
1347 if not isinstance(self.compress, str):
1348 self.compress = "deflate"
1349 self.headers[hdrs.CONTENT_ENCODING] = self.compress
1350 self.chunked = True # enable chunked, no need to deal with length
1352 def update_transfer_encoding(self) -> None:
1353 """Analyze transfer-encoding header."""
1354 te = self.headers.get(hdrs.TRANSFER_ENCODING, "").lower()
1356 if "chunked" in te:
1357 if self.chunked:
1358 raise ValueError(
1359 "chunked can not be set "
1360 'if "Transfer-Encoding: chunked" header is set'
1361 )
1363 elif self.chunked:
1364 if hdrs.CONTENT_LENGTH in self.headers:
1365 raise ValueError(
1366 "chunked can not be set if Content-Length header is set"
1367 )
1369 self.headers[hdrs.TRANSFER_ENCODING] = "chunked"
1371 def update_auth(self, auth: BasicAuth | None, trust_env: bool = False) -> None:
1372 """Set basic auth."""
1373 if auth is None:
1374 auth = self.auth
1375 if auth is None:
1376 return
1378 if not isinstance(auth, helpers.BasicAuth):
1379 raise TypeError("BasicAuth() tuple is required instead")
1381 self.headers[hdrs.AUTHORIZATION] = auth.encode()
1383 def update_body_from_data(self, body: Any, _stacklevel: int = 3) -> None:
1384 """Update request body from data."""
1385 if self._body is not None:
1386 _warn_if_unclosed_payload(self._body, stacklevel=_stacklevel)
1388 if body is None:
1389 self._body = None
1390 # Set Content-Length to 0 when body is None for methods that expect a body
1391 if (
1392 self.method not in self.GET_METHODS
1393 and not self.chunked
1394 and hdrs.CONTENT_LENGTH not in self.headers
1395 ):
1396 self.headers[hdrs.CONTENT_LENGTH] = "0"
1397 return
1399 # FormData
1400 maybe_payload = body() if isinstance(body, FormData) else body
1402 try:
1403 body_payload = payload.PAYLOAD_REGISTRY.get(maybe_payload, disposition=None)
1404 except payload.LookupError:
1405 body_payload = FormData(maybe_payload)() # type: ignore[arg-type]
1407 self._body = body_payload
1408 # enable chunked encoding if needed
1409 if not self.chunked and hdrs.CONTENT_LENGTH not in self.headers:
1410 if (size := body_payload.size) is not None:
1411 self.headers[hdrs.CONTENT_LENGTH] = str(size)
1412 else:
1413 self.chunked = True
1415 # copy payload headers
1416 assert body_payload.headers
1417 headers = self.headers
1418 skip_headers = self._skip_auto_headers
1419 for key, value in body_payload.headers.items():
1420 if key in headers or (skip_headers is not None and key in skip_headers):
1421 continue
1422 headers[key] = value
1424 def _update_body(self, body: Any) -> None:
1425 """Update request body after its already been set."""
1426 # Remove existing Content-Length header since body is changing
1427 if hdrs.CONTENT_LENGTH in self.headers:
1428 del self.headers[hdrs.CONTENT_LENGTH]
1430 # Remove existing Transfer-Encoding header to avoid conflicts
1431 if self.chunked and hdrs.TRANSFER_ENCODING in self.headers:
1432 del self.headers[hdrs.TRANSFER_ENCODING]
1434 # Now update the body using the existing method
1435 # Called from _update_body, add 1 to stacklevel from caller
1436 self.update_body_from_data(body, _stacklevel=4)
1438 # Update transfer encoding headers if needed (same logic as __init__)
1439 if body is not None or self.method not in self.GET_METHODS:
1440 self.update_transfer_encoding()
1442 async def update_body(self, body: Any) -> None:
1443 """
1444 Update request body and close previous payload if needed.
1446 This method safely updates the request body by first closing any existing
1447 payload to prevent resource leaks, then setting the new body.
1449 IMPORTANT: Always use this method instead of setting request.body directly.
1450 Direct assignment to request.body will leak resources if the previous body
1451 contains file handles, streams, or other resources that need cleanup.
1453 Args:
1454 body: The new body content. Can be:
1455 - bytes/bytearray: Raw binary data
1456 - str: Text data (will be encoded using charset from Content-Type)
1457 - FormData: Form data that will be encoded as multipart/form-data
1458 - Payload: A pre-configured payload object
1459 - AsyncIterable: An async iterable of bytes chunks
1460 - File-like object: Will be read and sent as binary data
1461 - None: Clears the body
1463 Usage:
1464 # CORRECT: Use update_body
1465 await request.update_body(b"new request data")
1467 # WRONG: Don't set body directly
1468 # request.body = b"new request data" # This will leak resources!
1470 # Update with form data
1471 form_data = FormData()
1472 form_data.add_field('field', 'value')
1473 await request.update_body(form_data)
1475 # Clear body
1476 await request.update_body(None)
1478 Note:
1479 This method is async because it may need to close file handles or
1480 other resources associated with the previous payload. Always await
1481 this method to ensure proper cleanup.
1483 Warning:
1484 Setting request.body directly is highly discouraged and can lead to:
1485 - Resource leaks (unclosed file handles, streams)
1486 - Memory leaks (unreleased buffers)
1487 - Unexpected behavior with streaming payloads
1489 It is not recommended to change the payload type in middleware. If the
1490 body was already set (e.g., as bytes), it's best to keep the same type
1491 rather than converting it (e.g., to str) as this may result in unexpected
1492 behavior.
1494 See Also:
1495 - update_body_from_data: Synchronous body update without cleanup
1496 - body property: Direct body access (STRONGLY DISCOURAGED)
1498 """
1499 # Close existing payload if it exists and needs closing
1500 if self._body is not None:
1501 await self._body.close()
1502 self._update_body(body)
1504 def update_expect_continue(self, expect: bool = False) -> None:
1505 if expect:
1506 self.headers[hdrs.EXPECT] = "100-continue"
1507 elif (
1508 hdrs.EXPECT in self.headers
1509 and self.headers[hdrs.EXPECT].lower() == "100-continue"
1510 ):
1511 expect = True
1513 if expect:
1514 self._continue = self.loop.create_future()
1516 def update_proxy(
1517 self,
1518 proxy: URL | None,
1519 proxy_auth: BasicAuth | None,
1520 proxy_headers: LooseHeaders | None,
1521 ) -> None:
1522 self.proxy = proxy
1523 if proxy is None:
1524 self.proxy_auth = None
1525 self.proxy_headers = None
1526 return
1528 if proxy_auth and not isinstance(proxy_auth, helpers.BasicAuth):
1529 raise ValueError("proxy_auth must be None or BasicAuth() tuple")
1530 self.proxy_auth = proxy_auth
1532 if proxy_headers is not None and not isinstance(
1533 proxy_headers, (MultiDict, MultiDictProxy)
1534 ):
1535 proxy_headers = CIMultiDict(proxy_headers)
1536 self.proxy_headers = proxy_headers
1538 async def write_bytes(
1539 self,
1540 writer: AbstractStreamWriter,
1541 conn: "Connection",
1542 content_length: int | None = None,
1543 ) -> None:
1544 """
1545 Write the request body to the connection stream.
1547 This method handles writing different types of request bodies:
1548 1. Payload objects (using their specialized write_with_length method)
1549 2. Bytes/bytearray objects
1550 3. Iterable body content
1552 Args:
1553 writer: The stream writer to write the body to
1554 conn: The connection being used for this request
1555 content_length: Optional maximum number of bytes to write from the body
1556 (None means write the entire body)
1558 The method properly handles:
1559 - Waiting for 100-Continue responses if required
1560 - Content length constraints for chunked encoding
1561 - Error handling for network issues, cancellation, and other exceptions
1562 - Signaling EOF and timeout management
1563 - Upload progress reporting to the request's UploadTracker
1565 Raises:
1566 ClientOSError: When there's an OS-level error writing the body
1567 ClientConnectionError: When there's a general connection error
1568 asyncio.CancelledError: When the operation is cancelled
1570 """
1571 tracker = self._upload_tracker
1572 gen = self._upload_gen
1573 if tracker is not None:
1574 writer.on_body_write = functools.partial(tracker._add_bytes, gen)
1575 tracker._attempt_writing(gen, writer)
1576 try:
1577 # 100 response
1578 if self._continue is not None:
1579 # Force headers to be sent before waiting for 100-continue
1580 writer.send_headers()
1581 await writer.drain()
1582 await self._continue
1584 protocol = conn.protocol
1585 assert protocol is not None
1586 try:
1587 # This should be a rare case but the
1588 # self._body can be set to None while
1589 # the task is being started or we wait above
1590 # for the 100-continue response.
1591 # The more likely case is we have an empty
1592 # payload, but 100-continue is still expected.
1593 if self._body is not None:
1594 await self._body.write_with_length(writer, content_length)
1595 except OSError as underlying_exc:
1596 reraised_exc = underlying_exc
1598 # Distinguish between timeout and other OS errors for better error reporting
1599 exc_is_not_timeout = underlying_exc.errno is not None or not isinstance(
1600 underlying_exc, asyncio.TimeoutError
1601 )
1602 if exc_is_not_timeout:
1603 reraised_exc = ClientOSError(
1604 underlying_exc.errno,
1605 f"Can not write request body for {self.url !s}",
1606 )
1608 set_exception(protocol, reraised_exc, underlying_exc)
1609 if tracker is not None:
1610 tracker._attempt_failed(gen, reraised_exc)
1611 except Exception as underlying_exc:
1612 wrapped_exc = ClientConnectionError(
1613 "Failed to send bytes into the underlying connection "
1614 f"{conn !s}: {underlying_exc!r}",
1615 )
1616 set_exception(protocol, wrapped_exc, underlying_exc)
1617 if tracker is not None:
1618 tracker._attempt_failed(gen, wrapped_exc)
1619 else:
1620 # Successfully wrote the body, signal EOF and start response timeout
1621 await writer.write_eof()
1622 if tracker is not None:
1623 tracker._attempt_finished(gen)
1624 protocol.start_timeout()
1625 except BaseException as underlying_exc:
1626 # Cancellation, or a failure escaping the inner handlers
1627 # (100-continue preamble, write_eof), leaves the body unsent:
1628 # the connection can't be reused, and the tracker must still
1629 # be notified. Record the attempt first: the tracker
1630 # transitions cannot fail, while conn.close() conceivably can.
1631 if tracker is not None:
1632 tracker._attempt_failed(
1633 gen,
1634 (
1635 None
1636 if isinstance(underlying_exc, asyncio.CancelledError)
1637 else underlying_exc
1638 ),
1639 )
1640 conn.close()
1641 raise
1643 async def send(self, conn: "Connection") -> "ClientResponse":
1644 # Specify request target:
1645 # - CONNECT request must send authority form URI
1646 # - not CONNECT proxy must send absolute form URI
1647 # - most common is origin form URI
1648 if self.method == hdrs.METH_CONNECT:
1649 connect_host = self.url.host_subcomponent
1650 assert connect_host is not None
1651 path = f"{connect_host}:{self.url.port}"
1652 elif self.proxy and not self.is_ssl():
1653 path = str(self.url)
1654 else:
1655 path = self.url.raw_path_qs
1657 protocol = conn.protocol
1658 assert protocol is not None
1659 writer = StreamWriter(
1660 protocol,
1661 self.loop,
1662 on_chunk_sent=(
1663 functools.partial(self._on_chunk_request_sent, self.method, self.url)
1664 if self._traces
1665 else None
1666 ),
1667 on_headers_sent=(
1668 functools.partial(self._on_headers_request_sent, self.method, self.url)
1669 if self._traces
1670 else None
1671 ),
1672 )
1674 if self.compress:
1675 writer.enable_compression(self.compress) # type: ignore[arg-type]
1677 if self.chunked is not None:
1678 writer.enable_chunking()
1680 # set default content-type
1681 if (
1682 self.method in self.POST_METHODS
1683 and (
1684 self._skip_auto_headers is None
1685 or hdrs.CONTENT_TYPE not in self._skip_auto_headers
1686 )
1687 and hdrs.CONTENT_TYPE not in self.headers
1688 ):
1689 self.headers[hdrs.CONTENT_TYPE] = "application/octet-stream"
1691 v = self.version
1692 if hdrs.CONNECTION not in self.headers:
1693 if conn._connector.force_close:
1694 if v == HttpVersion11:
1695 self.headers[hdrs.CONNECTION] = "close"
1696 elif v == HttpVersion10:
1697 self.headers[hdrs.CONNECTION] = "keep-alive"
1699 # status + headers
1700 status_line = f"{self.method} {path} HTTP/{v.major}.{v.minor}"
1702 # Buffer headers for potential coalescing with body
1703 await writer.write_headers(status_line, self.headers)
1705 task: asyncio.Task[None] | None
1706 if self._body or self._continue is not None or protocol.writing_paused:
1707 coro = self.write_bytes(writer, conn, self._get_content_length())
1708 if sys.version_info >= (3, 14):
1709 # Try to write bytes immediately to avoid having to schedule
1710 # the task on the event loop.
1711 loop = asyncio.get_running_loop()
1712 if isinstance(loop, BaseEventLoop):
1713 task = asyncio.create_task(coro, eager_start=True)
1714 else:
1715 task = asyncio.Task(coro, loop=loop, eager_start=True)
1716 elif sys.version_info >= (3, 12):
1717 task = asyncio.Task(
1718 coro, loop=asyncio.get_running_loop(), eager_start=True
1719 )
1720 else:
1721 task = asyncio.create_task(coro)
1722 if task.done():
1723 task = None
1724 else:
1725 self._writer = task
1726 else:
1727 # We have nothing to write because
1728 # - there is no body
1729 # - the protocol does not have writing paused
1730 # - we are not waiting for a 100-continue response
1731 protocol.start_timeout()
1732 writer.set_eof()
1733 task = None
1734 if (tracker := self._upload_tracker) is not None:
1735 # The request went out without a body-writer task: the
1736 # dispatched attempt is trivially complete with zero bytes.
1737 tracker._attempt_finished(self._upload_gen)
1738 response_class = self.response_class
1739 assert response_class is not None
1740 self.response = response_class(
1741 self.method,
1742 self.original_url,
1743 writer=task,
1744 continue100=self._continue,
1745 timer=self._timer,
1746 request_info=self.request_info,
1747 traces=self._traces,
1748 loop=self.loop,
1749 session=self._session,
1750 stream_writer=writer,
1751 )
1752 return self.response
1754 async def close(self) -> None:
1755 if self.__writer is not None:
1756 try:
1757 await self.__writer
1758 except asyncio.CancelledError:
1759 if (
1760 sys.version_info >= (3, 11)
1761 and (task := asyncio.current_task())
1762 and task.cancelling()
1763 ):
1764 raise
1766 def terminate(self) -> None:
1767 if self.__writer is not None:
1768 if not self.loop.is_closed():
1769 self.__writer.cancel()
1770 self.__writer.remove_done_callback(self.__reset_writer)
1771 self.__writer = None
1773 async def _on_chunk_request_sent(self, method: str, url: URL, chunk: bytes) -> None:
1774 for trace in self._traces:
1775 await trace.send_request_chunk_sent(method, url, chunk)
1777 async def _on_headers_request_sent(
1778 self, method: str, url: URL, headers: "CIMultiDict[str]"
1779 ) -> None:
1780 for trace in self._traces:
1781 await trace.send_request_headers(method, url, headers)