Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/client_reqrep.py: 34%
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 collections.abc import Callable, Iterable, Sequence
11from hashlib import md5, sha1, sha256
12from http.cookies import BaseCookie, SimpleCookie
13from types import MappingProxyType, TracebackType
14from typing import TYPE_CHECKING, Any, Literal, NamedTuple, TypedDict
16from multidict import CIMultiDict, CIMultiDictProxy, MultiDict, MultiDictProxy
17from yarl import URL, Query
19from . import hdrs, multipart, payload
20from ._cookie_helpers import (
21 parse_cookie_header,
22 parse_set_cookie_headers,
23 preserve_morsel_with_coded_value,
24)
25from .abc import AbstractStreamWriter
26from .base_protocol import BaseProtocol
27from .client_exceptions import (
28 ClientConnectionError,
29 ClientOSError,
30 ClientResponseError,
31 ContentTypeError,
32 InvalidURL,
33 ServerFingerprintMismatch,
34)
35from .compression_utils import HAS_BROTLI, HAS_ZSTD
36from .formdata import FormData
37from .helpers import (
38 _SENTINEL,
39 HTTP_AND_EMPTY_SCHEMA_SET,
40 BaseTimerContext,
41 HeadersDictProxy,
42 HeadersMixin,
43 TimerNoop,
44 encode_basic_auth,
45 frozen_dataclass_decorator,
46 is_expected_content_type,
47 parse_mimetype,
48 reify,
49 sentinel,
50 set_exception,
51 set_result,
52)
53from .http import (
54 SERVER_SOFTWARE,
55 HttpProcessingError,
56 HttpVersion,
57 HttpVersion10,
58 HttpVersion11,
59 StreamWriter,
60)
61from .streams import EMPTY_PAYLOAD, StreamReader
62from .typedefs import DEFAULT_JSON_DECODER, JSONDecoder, RawHeaders
64try:
65 import ssl
66 from ssl import SSLContext
67except ImportError: # pragma: no cover
68 ssl = None # type: ignore[assignment]
69 SSLContext = object # type: ignore[misc,assignment]
72__all__ = ("ClientRequest", "ClientResponse", "RequestInfo", "Fingerprint")
75if TYPE_CHECKING:
76 from .client import ClientSession
77 from .connector import Connection
78 from .tracing import Trace
81_CONNECTION_CLOSED_EXCEPTION = ClientConnectionError("Connection closed")
82_CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
83_DIGITS_RE = re.compile(r"\d+", re.ASCII)
86@frozen_dataclass_decorator
87class ClientTimeout:
88 total: float | None = 5 * 60 # 5 minute default timeout
89 connect: float | None = None
90 sock_read: float | None = None
91 sock_connect: float | None = None
92 ceil_threshold: float = 5
94 # pool_queue_timeout: Optional[float] = None
95 # dns_resolution_timeout: Optional[float] = None
96 # socket_connect_timeout: Optional[float] = None
97 # connection_acquiring_timeout: Optional[float] = None
98 # new_connection_timeout: Optional[float] = None
99 # http_header_timeout: Optional[float] = None
100 # response_body_timeout: Optional[float] = None
102 # to create a timeout specific for a single request, either
103 # - create a completely new one to overwrite the default
104 # - or use https://docs.python.org/3/library/dataclasses.html#dataclasses.replace
105 # to overwrite the defaults
107 def __post_init__(self) -> None:
108 # Ensure total is never lower than a more specific timeout, otherwise
109 # the latter would be silently capped by total and rendered useless.
110 # total=None means the user explicitly disabled the total timeout.
111 if self.total is None:
112 return
113 object.__setattr__(
114 self,
115 "total",
116 max(
117 self.total,
118 self.connect or 0,
119 self.sock_read or 0,
120 self.sock_connect or 0,
121 ),
122 )
124 if self.total == 0:
125 raise ValueError(
126 "total timeout must be a positive number or None to disable, "
127 "got 0. Using 0 to disable timeouts is no longer supported, "
128 "use None instead."
129 )
132def _gen_default_accept_encoding() -> str:
133 encodings = [
134 "gzip",
135 "deflate",
136 ]
137 if HAS_BROTLI:
138 encodings.append("br")
139 if HAS_ZSTD:
140 encodings.append("zstd")
141 return ", ".join(encodings)
144@frozen_dataclass_decorator
145class ContentDisposition:
146 type: str | None
147 parameters: "MappingProxyType[str, str]"
148 filename: str | None
151class _RequestInfo(NamedTuple):
152 url: URL
153 method: str
154 headers: "CIMultiDictProxy[str]"
155 real_url: URL
158class RequestInfo(_RequestInfo):
160 def __new__(
161 cls,
162 url: URL,
163 method: str,
164 headers: "CIMultiDictProxy[str]",
165 real_url: URL | _SENTINEL = sentinel,
166 ) -> "RequestInfo":
167 """Create a new RequestInfo instance.
169 For backwards compatibility, the real_url parameter is optional.
170 """
171 return tuple.__new__(
172 cls, (url, method, headers, url if real_url is sentinel else real_url)
173 )
176class Fingerprint:
177 HASHFUNC_BY_DIGESTLEN = {
178 16: md5,
179 20: sha1,
180 32: sha256,
181 }
183 def __init__(self, fingerprint: bytes) -> None:
184 digestlen = len(fingerprint)
185 hashfunc = self.HASHFUNC_BY_DIGESTLEN.get(digestlen)
186 if not hashfunc:
187 raise ValueError("fingerprint has invalid length")
188 elif hashfunc is md5 or hashfunc is sha1:
189 raise ValueError("md5 and sha1 are insecure and not supported. Use sha256.")
190 self._hashfunc = hashfunc
191 self._fingerprint = fingerprint
193 @property
194 def fingerprint(self) -> bytes:
195 return self._fingerprint
197 def check(self, transport: asyncio.Transport) -> None:
198 if not transport.get_extra_info("sslcontext"):
199 return
200 sslobj = transport.get_extra_info("ssl_object")
201 cert = sslobj.getpeercert(binary_form=True)
202 got = self._hashfunc(cert).digest()
203 if got != self._fingerprint:
204 host, port, *_ = transport.get_extra_info("peername")
205 raise ServerFingerprintMismatch(self._fingerprint, got, host, port)
208if ssl is not None:
209 SSL_ALLOWED_TYPES = (ssl.SSLContext, bool, Fingerprint)
210else: # pragma: no cover
211 SSL_ALLOWED_TYPES = (bool,) # type: ignore[unreachable]
214_CONNECTION_CLOSED_EXCEPTION = ClientConnectionError("Connection closed")
215_SSL_SCHEMES = frozenset(("https", "wss"))
218# ConnectionKey is a NamedTuple because it is used as a key in a dict
219# and a set in the connector. Since a NamedTuple is a tuple it uses
220# the fast native tuple __hash__ and __eq__ implementation in CPython.
221class ConnectionKey(NamedTuple):
222 # the key should contain an information about used proxy / TLS
223 # to prevent reusing wrong connections from a pool
224 host: str
225 port: int | None
226 is_ssl: bool
227 ssl: SSLContext | bool | Fingerprint
228 proxy: URL | None
229 proxy_headers_hash: int | None # hash(CIMultiDict)
230 server_hostname: str | None = None
233class ResponseParams(TypedDict):
234 timer: BaseTimerContext | None
235 skip_payload: bool
236 read_until_eof: bool
237 auto_decompress: bool
238 read_timeout: float | None
239 read_bufsize: int
240 timeout_ceil_threshold: float
241 max_line_size: int
242 max_field_size: int
243 max_headers: int
246class ClientResponse(HeadersMixin):
247 # Some of these attributes are None when created,
248 # but will be set by the start() method.
249 # As the end user will likely never see the None values, we cheat the types below.
250 # from the Status-Line of the response
251 version: HttpVersion | None = None # HTTP-Version
252 status: int = None # type: ignore[assignment] # Status-Code
253 reason: str | None = None # Reason-Phrase
255 content: StreamReader = None # type: ignore[assignment] # Payload stream
256 _body: bytes | None = None
257 _headers: HeadersDictProxy = None # type: ignore[assignment]
258 _history: tuple["ClientResponse", ...] = ()
259 _raw_headers: RawHeaders = None # type: ignore[assignment]
260 _upgraded: bool = False # parser saw a Connection: upgrade token
262 _connection: "Connection | None" = None # current connection
263 _cookies: SimpleCookie | None = None
264 _raw_cookie_headers: tuple[str, ...] | None = None
265 _continue: asyncio.Future[bool] | None = None
266 _source_traceback: traceback.StackSummary | None = None
267 _session: "ClientSession | None" = None
268 # set up by ClientRequest after ClientResponse object creation
269 # post-init stage allows to not change ctor signature
270 _closed = True # to allow __del__ for non-initialized properly response
271 _released = False
272 _in_context = False
274 _resolve_charset: Callable[["ClientResponse", bytes], str] = lambda *_: "utf-8"
276 __writer: asyncio.Task[None] | None = None
277 _stream_writer: AbstractStreamWriter | None = None
278 _output_size: int = 0
279 _upload_complete: asyncio.Future[None] | None = None
281 def __init__(
282 self,
283 method: str,
284 url: URL,
285 *,
286 writer: asyncio.Task[None] | None,
287 continue100: asyncio.Future[bool] | None,
288 timer: BaseTimerContext | None,
289 traces: Sequence["Trace"],
290 loop: asyncio.AbstractEventLoop,
291 session: "ClientSession | None",
292 request_headers: CIMultiDict[str],
293 original_url: URL,
294 stream_writer: AbstractStreamWriter,
295 **kwargs: object,
296 ) -> None:
297 # kwargs exists so authors of subclasses should expect to pass through unknown
298 # arguments. This allows us to safely add new arguments in future releases.
299 # But, we should never receive unknown arguments here in the parent class, this
300 # would indicate an argument has been named wrong or similar in the subclass.
301 assert not kwargs, "Unexpected arguments to ClientResponse"
302 # URL forbids subclasses, so a simple type check is enough.
303 assert type(url) is URL
305 self.method = method
307 self._real_url = url
308 self._url = url.with_fragment(None) if url.raw_fragment else url
309 if writer is None: # Request already sent
310 self._output_size = stream_writer.output_size
311 else:
312 self._stream_writer = stream_writer
313 self._writer = writer
314 if continue100 is not None:
315 self._continue = continue100
316 self._request_headers = request_headers
317 self._original_url = original_url
318 self._timer = timer if timer is not None else TimerNoop()
319 self._cache: dict[str, Any] = {}
320 self._traces = traces
321 self._loop = loop
322 # Save reference to _resolve_charset, so that get_encoding() will still
323 # work after the response has finished reading the body.
324 if session is not None:
325 # store a reference to session #1985
326 self._session = session
327 self._resolve_charset = session._resolve_charset
328 if loop.get_debug():
329 self._source_traceback = traceback.extract_stack(sys._getframe(1))
331 def __reset_writer(self, _: object = None) -> None:
332 self.__writer = None
333 if self._stream_writer is not None:
334 self._output_size = self._stream_writer.output_size
335 self._stream_writer = None
336 if self._upload_complete is not None and not self._upload_complete.done():
337 self._upload_complete.set_result(None)
339 @property
340 def _writer(self) -> asyncio.Task[None] | None:
341 """The writer task for streaming data.
343 _writer is only provided for backwards compatibility
344 for subclasses that may need to access it.
345 """
346 return self.__writer
348 @_writer.setter
349 def _writer(self, writer: asyncio.Task[None] | None) -> None:
350 """Set the writer task for streaming data."""
351 if self.__writer is not None:
352 self.__writer.remove_done_callback(self.__reset_writer)
353 self.__writer = writer
354 if writer is None:
355 return
356 if writer.done():
357 # The writer is already done, so we can clear it immediately.
358 self.__reset_writer()
359 else:
360 writer.add_done_callback(self.__reset_writer)
362 @property
363 def output_size(self) -> int:
364 """Number of bytes sent for this request."""
365 if self._stream_writer is not None:
366 return self._stream_writer.output_size
367 return self._output_size
369 @property
370 def upload_complete(self) -> "asyncio.Future[None]":
371 """Future set when the request body has been fully sent.
373 Already done when the request had no body or was written eagerly.
374 """
375 if self._upload_complete is None:
376 self._upload_complete = self._loop.create_future()
377 if self._stream_writer is None: # upload already finished
378 self._upload_complete.set_result(None)
379 return self._upload_complete
381 @property
382 def cookies(self) -> SimpleCookie:
383 if self._cookies is None:
384 if self._raw_cookie_headers is not None:
385 # Parse cookies for response.cookies (SimpleCookie for backward compatibility)
386 cookies = SimpleCookie()
387 # Use parse_set_cookie_headers for more lenient parsing that handles
388 # malformed cookies better than SimpleCookie.load
389 cookies.update(parse_set_cookie_headers(self._raw_cookie_headers))
390 self._cookies = cookies
391 else:
392 self._cookies = SimpleCookie()
393 return self._cookies
395 @cookies.setter
396 def cookies(self, cookies: SimpleCookie) -> None:
397 self._cookies = cookies
398 # Generate raw cookie headers from the SimpleCookie
399 if cookies:
400 self._raw_cookie_headers = tuple(
401 morsel.OutputString() for morsel in cookies.values()
402 )
403 else:
404 self._raw_cookie_headers = None
406 @reify
407 def url(self) -> URL:
408 return self._url
410 @reify
411 def real_url(self) -> URL:
412 return self._real_url
414 @reify
415 def host(self) -> str:
416 assert self._url.host is not None
417 return self._url.host
419 @reify
420 def headers(self) -> HeadersDictProxy:
421 return self._headers
423 @reify
424 def raw_headers(self) -> RawHeaders:
425 return self._raw_headers
427 @reify
428 def request_info(self) -> RequestInfo:
429 # Build RequestInfo lazily from components
430 headers = CIMultiDictProxy(self._request_headers)
431 return tuple.__new__(
432 RequestInfo, (self._url, self.method, headers, self._original_url)
433 )
435 @reify
436 def content_disposition(self) -> ContentDisposition | None:
437 raw = self._headers.get(hdrs.CONTENT_DISPOSITION)
438 if raw is None:
439 return None
440 disposition_type, params_dct = multipart.parse_content_disposition(raw)
441 params = MappingProxyType(params_dct)
442 filename = multipart.content_disposition_filename(params)
443 return ContentDisposition(disposition_type, params, filename)
445 def __del__(self, _warnings: Any = warnings) -> None:
446 if self._closed:
447 return
449 if self._connection is not None:
450 self._connection.release()
451 self._cleanup_writer()
453 if self._loop.get_debug():
454 _warnings.warn(
455 f"Unclosed response {self!r}", ResourceWarning, source=self
456 )
457 context = {"client_response": self, "message": "Unclosed response"}
458 if self._source_traceback:
459 context["source_traceback"] = self._source_traceback
460 self._loop.call_exception_handler(context)
462 def __repr__(self) -> str:
463 out = io.StringIO()
464 ascii_encodable_url = str(self.url)
465 if self.reason:
466 ascii_encodable_reason = self.reason.encode(
467 "ascii", "backslashreplace"
468 ).decode("ascii")
469 else:
470 ascii_encodable_reason = "None"
471 print(
472 f"<ClientResponse({ascii_encodable_url}) [{self.status} {ascii_encodable_reason}]>",
473 file=out,
474 )
475 print(self.headers, file=out)
476 return out.getvalue()
478 @property
479 def connection(self) -> "Connection | None":
480 return self._connection
482 @reify
483 def history(self) -> tuple["ClientResponse", ...]:
484 """A sequence of responses, if redirects occurred."""
485 return self._history
487 @reify
488 def links(self) -> "MultiDictProxy[MultiDictProxy[str | URL]]":
489 links: MultiDict[MultiDictProxy[str | URL]] = MultiDict()
490 for val in self.headers.getall("link"):
491 match = re.match(r"\s*<(.*)>(.*)", val)
492 if match is None: # Malformed link
493 continue
494 url, params_str = match.groups()
495 params = params_str.split(";")[1:]
497 link: MultiDict[str | URL] = MultiDict()
499 for param in params:
500 match = re.match(r"^\s*(\S*)\s*=\s*(['\"]?)(.*?)(\2)\s*$", param, re.M)
501 if match is None: # Malformed param
502 continue
503 key, _, value, _ = match.groups()
505 link.add(key, value)
507 key = link.get("rel", url)
509 link.add("url", self.url.join(URL(url)))
511 links.add(str(key), MultiDictProxy(link))
513 return MultiDictProxy(links)
515 async def start(self, connection: "Connection") -> "ClientResponse":
516 """Start response processing."""
517 self._closed = False
518 self._protocol = connection.protocol
519 self._connection = connection
521 with self._timer:
522 while True:
523 # read response
524 try:
525 protocol = self._protocol
526 message, payload = await protocol.read() # type: ignore[union-attr]
527 except HttpProcessingError as exc:
528 raise ClientResponseError(
529 self.request_info,
530 self.history,
531 status=exc.code,
532 message=exc.message,
533 headers=exc.headers,
534 ) from exc
536 if message.code < 100 or message.code > 199 or message.code == 101:
537 break
539 if self._continue is not None:
540 set_result(self._continue, True)
541 self._continue = None
543 # payload eof handler
544 payload.on_eof(self._response_eof)
546 # response status
547 self.version = message.version
548 self.status = message.code
549 self.reason = message.reason
551 # headers
552 self._headers = message.headers
553 self._raw_headers = message.raw_headers
554 self._upgraded = message.upgrade
556 # payload
557 self.content = payload
559 if self._traces and payload is not EMPTY_PAYLOAD:
560 payload._on_chunk_received = self._on_chunk_response_received
562 # cookies
563 if cookie_hdrs := self.headers._md.getall(hdrs.SET_COOKIE, ()):
564 # Store raw cookie headers for CookieJar
565 self._raw_cookie_headers = tuple(cookie_hdrs)
566 return self
568 def _response_eof(self) -> None:
569 if self._closed:
570 return
572 # protocol could be None because connection could be detached
573 protocol = self._connection and self._connection.protocol
574 if protocol is not None and protocol.upgraded:
575 return
577 self._closed = True
578 self._cleanup_writer()
579 self._release_connection()
581 @property
582 def closed(self) -> bool:
583 return self._closed
585 def close(self) -> None:
586 if not self._released:
587 self._notify_content()
589 self._closed = True
590 if self._loop.is_closed():
591 return
593 self._cleanup_writer()
594 if self._connection is not None:
595 self._connection.close()
596 self._connection = None
598 def release(self) -> None:
599 if not self._released:
600 self._notify_content()
602 self._closed = True
604 self._cleanup_writer()
605 self._release_connection()
607 @property
608 def ok(self) -> bool:
609 """Returns ``True`` if ``status`` is less than ``400``, ``False`` if not.
611 This is **not** a check for ``200 OK`` but a check that the response
612 status is under 400.
613 """
614 return 400 > self.status
616 def raise_for_status(self) -> None:
617 if not self.ok:
618 # reason should always be not None for a started response
619 assert self.reason is not None
621 # If we're in a context we can rely on __aexit__() to release as the
622 # exception propagates.
623 if not self._in_context:
624 self.release()
626 raise ClientResponseError(
627 self.request_info,
628 self.history,
629 status=self.status,
630 message=self.reason,
631 headers=self.headers,
632 )
634 def _release_connection(self) -> None:
635 if self._connection is not None:
636 if self.__writer is None:
637 self._connection.release()
638 self._connection = None
639 else:
640 self.__writer.add_done_callback(lambda f: self._release_connection())
642 async def _wait_released(self) -> None:
643 if self.__writer is not None:
644 try:
645 await self.__writer
646 except asyncio.CancelledError:
647 if (
648 sys.version_info >= (3, 11)
649 and (task := asyncio.current_task())
650 and task.cancelling()
651 ):
652 raise
653 self._release_connection()
655 def _cleanup_writer(self) -> None:
656 if self.__writer is not None:
657 self.__writer.cancel()
658 if self._stream_writer is not None:
659 self._output_size = self._stream_writer.output_size
660 self._stream_writer = None
661 self._session = None
663 def _notify_content(self) -> None:
664 content = self.content
665 # content can be None here, but the types are cheated elsewhere.
666 if content: # type: ignore[truthy-bool]
667 if content.exception() is None:
668 set_exception(content, _CONNECTION_CLOSED_EXCEPTION)
669 # The bound method installed in start() captures self, creating a
670 # response→payload→method→self cycle. Clear it eagerly so the
671 # response is reclaimable without waiting for cycle GC.
672 if content._on_chunk_received is not None:
673 content._on_chunk_received = None
674 self._released = True
676 async def wait_for_close(self) -> None:
677 if self.__writer is not None:
678 try:
679 await self.__writer
680 except asyncio.CancelledError:
681 if (
682 sys.version_info >= (3, 11)
683 and (task := asyncio.current_task())
684 and task.cancelling()
685 ):
686 raise
687 self.release()
689 async def _on_chunk_response_received(self, chunk: bytes) -> None:
690 try:
691 for trace in self._traces:
692 await trace.send_response_chunk_received(self.method, self.url, chunk)
693 except BaseException:
694 self.close()
695 raise
697 async def read(self) -> bytes:
698 """Read response payload."""
699 if self._body is None:
700 try:
701 self._body = await self.content.read()
702 except BaseException:
703 self.close()
704 raise
705 elif self._released: # Response explicitly released
706 raise ClientConnectionError("Connection closed")
708 protocol = self._connection and self._connection.protocol
709 if protocol is None or not protocol.upgraded:
710 await self._wait_released() # Underlying connection released
711 return self._body
713 def get_encoding(self) -> str:
714 ctype = self.headers.get(hdrs.CONTENT_TYPE, "").lower()
715 mimetype = parse_mimetype(ctype)
717 encoding = mimetype.parameters.get("charset")
718 if encoding:
719 with contextlib.suppress(LookupError, ValueError):
720 return codecs.lookup(encoding).name
722 if mimetype.type == "application" and (
723 mimetype.subtype == "json" or mimetype.subtype == "rdap"
724 ):
725 # RFC 7159 states that the default encoding is UTF-8.
726 # RFC 7483 defines application/rdap+json
727 return "utf-8"
729 if self._body is None:
730 raise RuntimeError(
731 "Cannot compute fallback encoding of a not yet read body"
732 )
734 return self._resolve_charset(self, self._body)
736 async def text(self, encoding: str | None = None, errors: str = "strict") -> str:
737 """Read response payload and decode."""
738 await self.read()
740 if encoding is None:
741 encoding = self.get_encoding()
743 return self._body.decode(encoding, errors=errors) # type: ignore[union-attr]
745 async def json(
746 self,
747 *,
748 encoding: str | None = None,
749 loads: JSONDecoder = DEFAULT_JSON_DECODER,
750 content_type: str | None = "application/json",
751 ) -> Any:
752 """Read and decodes JSON response."""
753 await self.read()
755 if content_type:
756 if not is_expected_content_type(self.content_type, content_type):
757 raise ContentTypeError(
758 self.request_info,
759 self.history,
760 status=self.status,
761 message=(
762 "Attempt to decode JSON with "
763 "unexpected mimetype: %s" % self.content_type
764 ),
765 headers=self.headers,
766 )
768 if encoding is None:
769 encoding = self.get_encoding()
771 return loads(self._body.decode(encoding)) # type: ignore[union-attr]
773 async def __aenter__(self) -> "ClientResponse":
774 self._in_context = True
775 return self
777 async def __aexit__(
778 self,
779 exc_type: type[BaseException] | None,
780 exc_val: BaseException | None,
781 exc_tb: TracebackType | None,
782 ) -> None:
783 self._in_context = False
784 # similar to _RequestContextManager, we do not need to check
785 # for exceptions, response object can close connection
786 # if state is broken
787 self.release()
788 await self.wait_for_close()
791class ClientRequestBase:
792 """An internal class for proxy requests."""
794 POST_METHODS = {hdrs.METH_PATCH, hdrs.METH_POST, hdrs.METH_PUT}
796 proxy: URL | None = None
797 response_class = ClientResponse
798 server_hostname: str | None = None # Needed in connector.py
799 version = HttpVersion11
800 _response = None
802 # These class defaults help create_autospec() work correctly.
803 # If autospec is improved in future, maybe these can be removed.
804 url = URL()
805 method = "GET"
807 _writer_task: asyncio.Task[None] | None = None # async task for streaming data
809 _skip_auto_headers: "CIMultiDict[None] | None" = None
811 # N.B.
812 # Adding __del__ method with self._writer closing doesn't make sense
813 # because _writer is instance method, thus it keeps a reference to self.
814 # Until writer has finished finalizer will not be called.
816 def __init__(
817 self,
818 method: str,
819 url: URL,
820 *,
821 headers: CIMultiDict[str],
822 loop: asyncio.AbstractEventLoop,
823 ssl: SSLContext | bool | Fingerprint,
824 trust_env: bool = False,
825 ):
826 if match := _CONTAINS_CONTROL_CHAR_RE.search(method):
827 raise ValueError(
828 f"Method cannot contain non-token characters {method!r} "
829 f"(found at least {match.group()!r})"
830 )
831 # URL forbids subclasses, so a simple type check is enough.
832 assert type(url) is URL, url
833 self.original_url = url
834 self.url = url.with_fragment(None) if url.raw_fragment else url
835 self.method = method.upper()
836 self.loop = loop
837 self._ssl = ssl
839 if loop.get_debug():
840 self._source_traceback = traceback.extract_stack(sys._getframe(1))
842 if not url.raw_host:
843 raise InvalidURL(url)
844 self._update_headers(headers)
845 if url.raw_user or url.raw_password:
846 self.headers[hdrs.AUTHORIZATION] = encode_basic_auth(
847 url.user or "", url.password or ""
848 )
850 def _reset_writer(self, _: object = None) -> None:
851 self._writer_task = None
853 def _get_content_length(self) -> int | None:
854 """Extract and validate Content-Length header value.
856 Returns parsed Content-Length value or None if not set.
857 Raises ValueError if header exists but cannot be parsed as an integer.
858 """
859 if hdrs.CONTENT_LENGTH not in self.headers:
860 return None
862 content_length_hdr = self.headers[hdrs.CONTENT_LENGTH]
863 if not _DIGITS_RE.fullmatch(content_length_hdr):
864 raise ValueError(f"Invalid Content-Length header: {content_length_hdr!r}")
865 return int(content_length_hdr)
867 @property
868 def _writer(self) -> asyncio.Task[None] | None:
869 return self._writer_task
871 @_writer.setter
872 def _writer(self, writer: asyncio.Task[None]) -> None:
873 if self._writer_task is not None:
874 self._writer_task.remove_done_callback(self._reset_writer)
875 self._writer_task = writer
876 writer.add_done_callback(self._reset_writer)
878 def is_ssl(self) -> bool:
879 return self.url.scheme in _SSL_SCHEMES
881 @property
882 def ssl(self) -> "SSLContext | bool | Fingerprint":
883 return self._ssl
885 @property
886 def connection_key(self) -> ConnectionKey:
887 url = self.url
888 return tuple.__new__(
889 ConnectionKey,
890 (
891 url.raw_host or "",
892 url.port,
893 url.scheme in _SSL_SCHEMES,
894 self._ssl,
895 None,
896 None,
897 self.server_hostname,
898 ),
899 )
901 def _update_headers(self, headers: CIMultiDict[str]) -> None:
902 """Update request headers."""
903 self.headers: CIMultiDict[str] = CIMultiDict()
905 # Build the host header
906 host = self.url.host_port_subcomponent
908 # host_port_subcomponent is None when the URL is a relative URL.
909 # but we know we do not have a relative URL here.
910 assert host is not None
911 self.headers[hdrs.HOST] = headers.pop(hdrs.HOST, host)
912 self.headers.extend(headers)
914 def _create_response(
915 self,
916 task: asyncio.Task[None] | None,
917 stream_writer: AbstractStreamWriter,
918 ) -> ClientResponse:
919 return self.response_class(
920 self.method,
921 self.original_url,
922 writer=task,
923 continue100=None,
924 timer=TimerNoop(),
925 traces=(),
926 loop=self.loop,
927 session=None,
928 request_headers=self.headers,
929 original_url=self.original_url,
930 stream_writer=stream_writer,
931 )
933 def _create_writer(self, protocol: BaseProtocol) -> StreamWriter:
934 return StreamWriter(protocol, self.loop)
936 def _should_write(self, protocol: BaseProtocol) -> bool:
937 return protocol.writing_paused
939 async def _send(self, conn: "Connection") -> ClientResponse:
940 # Specify request target:
941 # - CONNECT request must send authority form URI
942 # - not CONNECT proxy must send absolute form URI
943 # - most common is origin form URI
944 if self.method == hdrs.METH_CONNECT:
945 connect_host = self.url.host_subcomponent
946 assert connect_host is not None
947 path = f"{connect_host}:{self.url.port}"
948 elif self.proxy and not self.is_ssl():
949 path = str(self.url)
950 else:
951 path = self.url.raw_path_qs
953 protocol = conn.protocol
954 assert protocol is not None
955 writer = self._create_writer(protocol)
957 # set default content-type
958 if (
959 self.method in self.POST_METHODS
960 and (
961 self._skip_auto_headers is None
962 or hdrs.CONTENT_TYPE not in self._skip_auto_headers
963 )
964 and hdrs.CONTENT_TYPE not in self.headers
965 ):
966 self.headers[hdrs.CONTENT_TYPE] = "application/octet-stream"
968 v = self.version
969 if hdrs.CONNECTION not in self.headers:
970 if conn._connector.force_close:
971 if v == HttpVersion11:
972 self.headers[hdrs.CONNECTION] = "close"
973 elif v == HttpVersion10:
974 self.headers[hdrs.CONNECTION] = "keep-alive"
976 # status + headers
977 status_line = f"{self.method} {path} HTTP/{v.major}.{v.minor}"
979 # Buffer headers for potential coalescing with body
980 await writer.write_headers(status_line, self.headers)
982 task: asyncio.Task[None] | None
983 if self._should_write(protocol):
984 coro = self._write_bytes(writer, conn, self._get_content_length())
985 if sys.version_info >= (3, 12):
986 # Optimization for Python 3.12, try to write
987 # bytes immediately to avoid having to schedule
988 # the task on the event loop.
989 task = asyncio.Task(coro, loop=self.loop, eager_start=True)
990 else:
991 task = self.loop.create_task(coro)
992 if task.done():
993 task = None
994 else:
995 self._writer = task
996 else:
997 # We have nothing to write because
998 # - there is no body
999 # - the protocol does not have writing paused
1000 # - we are not waiting for a 100-continue response
1001 protocol.start_timeout()
1002 writer.set_eof()
1003 task = None
1004 self._response = self._create_response(task, stream_writer=writer)
1005 return self._response
1007 async def _write_bytes(
1008 self,
1009 writer: AbstractStreamWriter,
1010 conn: "Connection",
1011 content_length: int | None,
1012 ) -> None:
1013 # Base class never has a body, this will never be run.
1014 assert False
1017class ClientRequestArgs(TypedDict, total=False):
1018 params: Query
1019 headers: CIMultiDict[str]
1020 skip_auto_headers: Iterable[str] | None
1021 data: Any
1022 cookies: BaseCookie[str]
1023 version: HttpVersion
1024 compress: Literal["deflate", "gzip"] | bool
1025 chunked: bool | None
1026 expect100: bool
1027 loop: asyncio.AbstractEventLoop
1028 response_class: type[ClientResponse]
1029 proxy: URL | None
1030 response_params: ResponseParams
1031 timer: BaseTimerContext
1032 timeout: ClientTimeout
1033 session: "ClientSession"
1034 ssl: SSLContext | bool | Fingerprint
1035 proxy_headers: CIMultiDict[str] | None
1036 traces: list["Trace"]
1037 trust_env: bool
1038 server_hostname: str | None
1041class ClientRequest(ClientRequestBase):
1042 _EMPTY_BODY = payload.PAYLOAD_REGISTRY.get(b"", disposition=None)
1043 _body = _EMPTY_BODY
1044 _continue = None # waiter future for '100 Continue' response
1045 _response_params: ResponseParams = None # type: ignore[assignment]
1046 _session: "ClientSession" = None # type: ignore[assignment]
1047 _timeout = ClientTimeout()
1048 _traces: list["Trace"] = () # type: ignore[assignment]
1050 GET_METHODS = {
1051 hdrs.METH_GET,
1052 hdrs.METH_HEAD,
1053 hdrs.METH_OPTIONS,
1054 hdrs.METH_TRACE,
1055 }
1056 DEFAULT_HEADERS = {
1057 hdrs.ACCEPT: "*/*",
1058 hdrs.ACCEPT_ENCODING: _gen_default_accept_encoding(),
1059 }
1061 def __init__(
1062 self,
1063 method: str,
1064 url: URL,
1065 *,
1066 params: Query,
1067 headers: CIMultiDict[str],
1068 skip_auto_headers: Iterable[str] | None,
1069 data: Any,
1070 cookies: BaseCookie[str],
1071 version: HttpVersion,
1072 compress: Literal["deflate", "gzip"] | bool,
1073 chunked: bool | None,
1074 expect100: bool,
1075 loop: asyncio.AbstractEventLoop,
1076 response_class: type[ClientResponse],
1077 proxy: URL | None,
1078 response_params: ResponseParams,
1079 timer: BaseTimerContext,
1080 timeout: ClientTimeout,
1081 session: "ClientSession",
1082 ssl: SSLContext | bool | Fingerprint,
1083 proxy_headers: CIMultiDict[str] | None,
1084 traces: list["Trace"],
1085 trust_env: bool,
1086 server_hostname: str | None,
1087 **kwargs: object,
1088 ):
1089 # kwargs exists so authors of subclasses should expect to pass through unknown
1090 # arguments. This allows us to safely add new arguments in future releases.
1091 # But, we should never receive unknown arguments here in the parent class, this
1092 # would indicate an argument has been named wrong or similar in the subclass.
1093 assert not kwargs, "Unexpected arguments to ClientRequest"
1095 if params:
1096 url = url.extend_query(params)
1097 super().__init__(method, url, headers=headers, loop=loop, ssl=ssl)
1099 if proxy is not None:
1100 assert type(proxy) is URL, proxy
1101 self._session = session
1102 self.chunked = chunked
1103 self.response_class = response_class
1104 self._response_params = response_params
1105 self._timer = timer
1106 self._timeout = timeout
1107 self.server_hostname = server_hostname
1108 self.version = version
1110 self._update_auto_headers(skip_auto_headers)
1111 self._update_cookies(cookies)
1112 self._update_content_encoding(data, compress)
1113 self._update_proxy(proxy, proxy_headers)
1115 self._update_body_from_data(data)
1116 if data is not None or self.method not in self.GET_METHODS:
1117 self._update_transfer_encoding()
1118 self._update_expect_continue(expect100)
1119 self._traces = traces
1121 @property
1122 def body(self) -> payload.Payload:
1123 return self._body
1125 @property
1126 def skip_auto_headers(self) -> CIMultiDict[None]:
1127 return self._skip_auto_headers or CIMultiDict()
1129 @property
1130 def timeout(self) -> ClientTimeout:
1131 """The timeout configuration this request runs under (read-only)."""
1132 return self._timeout
1134 @property
1135 def connection_key(self) -> ConnectionKey:
1136 if proxy_headers := self.proxy_headers:
1137 h: int | None = hash(tuple(proxy_headers.items()))
1138 else:
1139 h = None
1140 url = self.url
1141 return tuple.__new__(
1142 ConnectionKey,
1143 (
1144 url.raw_host or "",
1145 url.port,
1146 url.scheme in _SSL_SCHEMES,
1147 self._ssl,
1148 self.proxy,
1149 h,
1150 self.server_hostname,
1151 ),
1152 )
1154 @property
1155 def session(self) -> "ClientSession":
1156 """Return the ClientSession instance.
1158 This property provides access to the ClientSession that initiated
1159 this request, allowing middleware to make additional requests
1160 using the same session.
1161 """
1162 return self._session
1164 def _update_auto_headers(self, skip_auto_headers: Iterable[str] | None) -> None:
1165 if skip_auto_headers is not None:
1166 self._skip_auto_headers = CIMultiDict(
1167 (hdr, None) for hdr in sorted(skip_auto_headers)
1168 )
1169 used_headers = self.headers.copy()
1170 used_headers.extend(self._skip_auto_headers) # type: ignore[arg-type]
1171 else:
1172 # Fast path when there are no headers to skip
1173 # which is the most common case.
1174 used_headers = self.headers
1176 for hdr, val in self.DEFAULT_HEADERS.items():
1177 if hdr not in used_headers:
1178 self.headers[hdr] = val
1180 if hdrs.USER_AGENT not in used_headers:
1181 self.headers[hdrs.USER_AGENT] = SERVER_SOFTWARE
1183 def _update_cookies(self, cookies: BaseCookie[str]) -> None:
1184 """Update request cookies header."""
1185 if not cookies:
1186 return
1188 c = SimpleCookie()
1189 if hdrs.COOKIE in self.headers:
1190 # parse_cookie_header for RFC 6265 compliant Cookie header parsing
1191 c.update(parse_cookie_header(self.headers.get(hdrs.COOKIE, "")))
1192 del self.headers[hdrs.COOKIE]
1194 for name, value in cookies.items():
1195 # Use helper to preserve coded_value exactly as sent by server
1196 c[name] = preserve_morsel_with_coded_value(value)
1198 self.headers[hdrs.COOKIE] = c.output(header="", sep=";").strip()
1200 def _update_content_encoding(
1201 self, data: Any, compress: bool | Literal["deflate", "gzip"]
1202 ) -> None:
1203 """Set request content encoding."""
1204 self.compress = None
1205 if not data:
1206 return
1208 if self.headers.get(hdrs.CONTENT_ENCODING):
1209 if compress:
1210 raise ValueError(
1211 "compress can not be set if Content-Encoding header is set"
1212 )
1213 elif compress:
1214 if isinstance(compress, str) and compress not in {"deflate", "gzip"}:
1215 raise ValueError(
1216 "compress must be one of True, False, 'deflate', or 'gzip'"
1217 )
1218 self.compress = compress if isinstance(compress, str) else "deflate"
1219 self.headers[hdrs.CONTENT_ENCODING] = self.compress
1220 self.chunked = True # enable chunked, no need to deal with length
1222 def _update_transfer_encoding(self) -> None:
1223 """Analyze transfer-encoding header."""
1224 te = self.headers.get(hdrs.TRANSFER_ENCODING, "").lower()
1226 if "chunked" in te:
1227 if self.chunked:
1228 raise ValueError(
1229 "chunked can not be set "
1230 'if "Transfer-Encoding: chunked" header is set'
1231 )
1233 elif self.chunked:
1234 if hdrs.CONTENT_LENGTH in self.headers:
1235 raise ValueError(
1236 "chunked can not be set if Content-Length header is set"
1237 )
1239 self.headers[hdrs.TRANSFER_ENCODING] = "chunked"
1241 def _update_body_from_data(self, body: Any) -> None:
1242 """Update request body from data."""
1243 if body is None:
1244 self._body = self._EMPTY_BODY
1245 # Set Content-Length to 0 when body is None for methods that expect a body
1246 if (
1247 self.method not in self.GET_METHODS
1248 and not self.chunked
1249 and hdrs.CONTENT_LENGTH not in self.headers
1250 ):
1251 self.headers[hdrs.CONTENT_LENGTH] = "0"
1252 return
1254 # FormData
1255 if isinstance(body, FormData):
1256 body = body()
1257 else:
1258 try:
1259 body = payload.PAYLOAD_REGISTRY.get(body, disposition=None)
1260 except payload.LookupError:
1261 boundary = None
1262 if hdrs.CONTENT_TYPE in self.headers:
1263 boundary = parse_mimetype(
1264 self.headers[hdrs.CONTENT_TYPE]
1265 ).parameters.get("boundary")
1266 body = FormData(body, boundary=boundary)()
1268 self._body = body
1270 # enable chunked encoding if needed
1271 if not self.chunked and hdrs.CONTENT_LENGTH not in self.headers:
1272 if (size := body.size) is not None:
1273 self.headers[hdrs.CONTENT_LENGTH] = str(size)
1274 else:
1275 self.chunked = True
1277 # copy payload headers
1278 assert body.headers
1279 headers = self.headers
1280 skip_headers = self._skip_auto_headers
1281 for key, value in body.headers.items():
1282 if key in headers or (skip_headers is not None and key in skip_headers):
1283 continue
1284 headers[key] = value
1286 def _update_body(self, body: Any) -> None:
1287 """Update request body after its already been set."""
1288 # Remove existing Content-Length header since body is changing
1289 if hdrs.CONTENT_LENGTH in self.headers:
1290 del self.headers[hdrs.CONTENT_LENGTH]
1292 # Remove existing Transfer-Encoding header to avoid conflicts
1293 if self.chunked and hdrs.TRANSFER_ENCODING in self.headers:
1294 del self.headers[hdrs.TRANSFER_ENCODING]
1296 # Now update the body using the existing method
1297 self._update_body_from_data(body)
1299 # Update transfer encoding headers if needed (same logic as __init__)
1300 if body is not None or self.method not in self.GET_METHODS:
1301 self._update_transfer_encoding()
1303 async def update_body(self, body: Any) -> None:
1304 """
1305 Update request body and close previous payload if needed.
1307 This method safely updates the request body by first closing any existing
1308 payload to prevent resource leaks, then setting the new body.
1310 IMPORTANT: Always use this method instead of setting request.body directly.
1311 Direct assignment to request.body will leak resources if the previous body
1312 contains file handles, streams, or other resources that need cleanup.
1314 Args:
1315 body: The new body content. Can be:
1316 - bytes/bytearray: Raw binary data
1317 - str: Text data (will be encoded using charset from Content-Type)
1318 - FormData: Form data that will be encoded as multipart/form-data
1319 - Payload: A pre-configured payload object
1320 - AsyncIterable: An async iterable of bytes chunks
1321 - File-like object: Will be read and sent as binary data
1322 - None: Clears the body
1324 Usage:
1325 # CORRECT: Use update_body
1326 await request.update_body(b"new request data")
1328 # WRONG: Don't set body directly
1329 # request.body = b"new request data" # This will leak resources!
1331 # Update with form data
1332 form_data = FormData()
1333 form_data.add_field('field', 'value')
1334 await request.update_body(form_data)
1336 # Clear body
1337 await request.update_body(None)
1339 Note:
1340 This method is async because it may need to close file handles or
1341 other resources associated with the previous payload. Always await
1342 this method to ensure proper cleanup.
1344 Warning:
1345 Setting request.body directly is highly discouraged and can lead to:
1346 - Resource leaks (unclosed file handles, streams)
1347 - Memory leaks (unreleased buffers)
1348 - Unexpected behavior with streaming payloads
1350 It is not recommended to change the payload type in middleware. If the
1351 body was already set (e.g., as bytes), it's best to keep the same type
1352 rather than converting it (e.g., to str) as this may result in unexpected
1353 behavior.
1355 See Also:
1356 - update_body_from_data: Synchronous body update without cleanup
1357 - body property: Direct body access (STRONGLY DISCOURAGED)
1359 """
1360 # Close existing payload if it exists and needs closing
1361 if self._body is not None:
1362 await self._body.close()
1363 self._update_body(body)
1365 def _update_expect_continue(self, expect: bool = False) -> None:
1366 if expect:
1367 self.headers[hdrs.EXPECT] = "100-continue"
1368 elif (
1369 hdrs.EXPECT in self.headers
1370 and self.headers[hdrs.EXPECT].lower() == "100-continue"
1371 ):
1372 expect = True
1374 if expect:
1375 self._continue = self.loop.create_future()
1377 def _update_proxy(
1378 self,
1379 proxy: URL | None,
1380 proxy_headers: CIMultiDict[str] | None,
1381 ) -> None:
1382 if proxy is None:
1383 self.proxy = None
1384 self.proxy_headers = None
1385 return
1387 if proxy.scheme not in HTTP_AND_EMPTY_SCHEMA_SET:
1388 raise ValueError(
1389 f"aiohttp only supports http(s) proxies (got: {proxy.scheme!r}).\n"
1390 "See third-party libraries for other proxy schemes."
1391 )
1393 # URL-embedded credentials on the proxy map to Proxy-Authorization.
1394 if proxy.raw_user or proxy.raw_password:
1395 auth_header = encode_basic_auth(proxy.user or "", proxy.password or "")
1396 if proxy_headers is None:
1397 proxy_headers = CIMultiDict()
1398 proxy_headers.setdefault(hdrs.PROXY_AUTHORIZATION, auth_header)
1399 proxy = proxy.with_user(None)
1400 self.proxy = proxy
1401 self.proxy_headers = proxy_headers
1403 def _create_response(
1404 self,
1405 task: asyncio.Task[None] | None,
1406 stream_writer: AbstractStreamWriter,
1407 ) -> ClientResponse:
1408 return self.response_class(
1409 self.method,
1410 self.original_url,
1411 writer=task,
1412 continue100=self._continue,
1413 timer=self._timer,
1414 traces=self._traces,
1415 loop=self.loop,
1416 session=self._session,
1417 request_headers=self.headers,
1418 original_url=self.original_url,
1419 stream_writer=stream_writer,
1420 )
1422 def _create_writer(self, protocol: BaseProtocol) -> StreamWriter:
1423 writer = StreamWriter(
1424 protocol,
1425 self.loop,
1426 on_chunk_sent=(
1427 functools.partial(self._on_chunk_request_sent, self.method, self.url)
1428 if self._traces
1429 else None
1430 ),
1431 on_headers_sent=(
1432 functools.partial(self._on_headers_request_sent, self.method, self.url)
1433 if self._traces
1434 else None
1435 ),
1436 )
1438 if self.compress:
1439 writer.enable_compression(self.compress)
1441 if self.chunked is not None:
1442 writer.enable_chunking()
1443 return writer
1445 def _should_write(self, protocol: BaseProtocol) -> bool:
1446 return (
1447 self.body.size != 0 or self._continue is not None or protocol.writing_paused
1448 )
1450 async def _write_bytes(
1451 self,
1452 writer: AbstractStreamWriter,
1453 conn: "Connection",
1454 content_length: int | None,
1455 ) -> None:
1456 """
1457 Write the request body to the connection stream.
1459 This method handles writing different types of request bodies:
1460 1. Payload objects (using their specialized write_with_length method)
1461 2. Bytes/bytearray objects
1462 3. Iterable body content
1464 Args:
1465 writer: The stream writer to write the body to
1466 conn: The connection being used for this request
1467 content_length: Optional maximum number of bytes to write from the body
1468 (None means write the entire body)
1470 The method properly handles:
1471 - Waiting for 100-Continue responses if required
1472 - Content length constraints for chunked encoding
1473 - Error handling for network issues, cancellation, and other exceptions
1474 - Signaling EOF and timeout management
1476 Raises:
1477 ClientOSError: When there's an OS-level error writing the body
1478 ClientConnectionError: When there's a general connection error
1479 asyncio.CancelledError: When the operation is cancelled
1481 """
1482 # 100 response
1483 if self._continue is not None:
1484 # Force headers to be sent before waiting for 100-continue
1485 writer.send_headers()
1486 await writer.drain()
1487 await self._continue
1489 protocol = conn.protocol
1490 assert protocol is not None
1491 try:
1492 await self._body.write_with_length(writer, content_length)
1493 except OSError as underlying_exc:
1494 reraised_exc = underlying_exc
1496 # Distinguish between timeout and other OS errors for better error reporting
1497 exc_is_not_timeout = underlying_exc.errno is not None or not isinstance(
1498 underlying_exc, asyncio.TimeoutError
1499 )
1500 if exc_is_not_timeout:
1501 reraised_exc = ClientOSError(
1502 underlying_exc.errno,
1503 f"Can not write request body for {self.url !s}",
1504 )
1506 set_exception(protocol, reraised_exc, underlying_exc)
1507 except asyncio.CancelledError:
1508 # Body hasn't been fully sent, so connection can't be reused
1509 conn.close()
1510 raise
1511 except Exception as underlying_exc:
1512 set_exception(
1513 protocol,
1514 ClientConnectionError(
1515 "Failed to send bytes into the underlying connection "
1516 f"{conn !s}: {underlying_exc!r}",
1517 ),
1518 underlying_exc,
1519 )
1520 else:
1521 # Successfully wrote the body, signal EOF and start response timeout
1522 await writer.write_eof()
1523 protocol.start_timeout()
1525 async def _close(self) -> None:
1526 if self._writer_task is not None:
1527 try:
1528 await self._writer_task
1529 except asyncio.CancelledError:
1530 if (
1531 sys.version_info >= (3, 11)
1532 and (task := asyncio.current_task())
1533 and task.cancelling()
1534 ):
1535 raise
1537 def _terminate(self) -> None:
1538 if self._writer_task is not None:
1539 if not self.loop.is_closed():
1540 self._writer_task.cancel()
1541 self._writer_task.remove_done_callback(self._reset_writer)
1542 self._writer_task = None
1544 async def _on_chunk_request_sent(self, method: str, url: URL, chunk: bytes) -> None:
1545 for trace in self._traces:
1546 await trace.send_request_chunk_sent(method, url, chunk)
1548 async def _on_headers_request_sent(
1549 self, method: str, url: URL, headers: "CIMultiDict[str]"
1550 ) -> None:
1551 for trace in self._traces:
1552 await trace.send_request_headers(method, url, headers)