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