codekingpro/portable-devtools
115k
1"""HTTP Client for asyncio."""
2
3import asyncio
4import base64
5import hashlib
6import json
7import os
8import sys
9import traceback
10import warnings
11from contextlib import suppress
12from types import TracebackType
13from typing import (
14 TYPE_CHECKING,
15 Any,
16 Awaitable,
17 Callable,
18 Coroutine,
19 Final,
20 FrozenSet,
21 Generator,
22 Generic,
23 Iterable,
24 List,
25 Mapping,
26 Optional,
27 Sequence,
28 Set,
29 Tuple,
30 Type,
31 TypedDict,
32 TypeVar,
33 Union,
34)
35
36import attr
37from multidict import CIMultiDict, MultiDict, MultiDictProxy, istr
38from yarl import URL
39
40from . import hdrs, http, payload
41from ._websocket.reader import WebSocketDataQueue
42from .abc import AbstractCookieJar
43from .client_exceptions import (
44 ClientConnectionError,
45 ClientConnectionResetError,
46 ClientConnectorCertificateError,
47 ClientConnectorDNSError,
48 ClientConnectorError,
49 ClientConnectorSSLError,
50 ClientError,
51 ClientHttpProxyError,
52 ClientOSError,
53 ClientPayloadError,
54 ClientProxyConnectionError,
55 ClientResponseError,
56 ClientSSLError,
57 ConnectionTimeoutError,
58 ContentTypeError,
59 InvalidURL,
60 InvalidUrlClientError,
61 InvalidUrlRedirectClientError,
62 NonHttpUrlClientError,
63 NonHttpUrlRedirectClientError,
64 RedirectClientError,
65 ServerConnectionError,
66 ServerDisconnectedError,
67 ServerFingerprintMismatch,
68 ServerTimeoutError,
69 SocketTimeoutError,
70 TooManyRedirects,
71 WSMessageTypeError,
72 WSServerHandshakeError,
73)
74from .client_middlewares import ClientMiddlewareType, build_client_middlewares
75from .client_reqrep import (
76 ClientRequest as ClientRequest,
77 ClientResponse as ClientResponse,
78 Fingerprint as Fingerprint,
79 RequestInfo as RequestInfo,
80 _merge_ssl_params,
81)
82from .client_ws import (
83 DEFAULT_WS_CLIENT_TIMEOUT,
84 ClientWebSocketResponse as ClientWebSocketResponse,
85 ClientWSTimeout as ClientWSTimeout,
86)
87from .connector import (
88 HTTP_AND_EMPTY_SCHEMA_SET,
89 BaseConnector as BaseConnector,
90 NamedPipeConnector as NamedPipeConnector,
91 TCPConnector as TCPConnector,
92 UnixConnector as UnixConnector,
93)
94from .cookiejar import CookieJar
95from .helpers import (
96 _SENTINEL,
97 DEBUG,
98 EMPTY_BODY_METHODS,
99 BasicAuth,
100 TimeoutHandle,
101 basicauth_from_netrc,
102 get_env_proxy_for_url,
103 netrc_from_env,
104 sentinel,
105 strip_auth_from_url,
106)
107from .http import WS_KEY, HttpVersion, WebSocketReader, WebSocketWriter
108from .http_websocket import WSHandshakeError, ws_ext_gen, ws_ext_parse
109from .tracing import Trace, TraceConfig
110from .typedefs import JSONEncoder, LooseCookies, LooseHeaders, Query, StrOrURL
111
112__all__ = (
113 # client_exceptions
114 "ClientConnectionError",
115 "ClientConnectionResetError",
116 "ClientConnectorCertificateError",
117 "ClientConnectorDNSError",
118 "ClientConnectorError",
119 "ClientConnectorSSLError",
120 "ClientError",
121 "ClientHttpProxyError",
122 "ClientOSError",
123 "ClientPayloadError",
124 "ClientProxyConnectionError",
125 "ClientResponseError",
126 "ClientSSLError",
127 "ConnectionTimeoutError",
128 "ContentTypeError",
129 "InvalidURL",
130 "InvalidUrlClientError",
131 "RedirectClientError",
132 "NonHttpUrlClientError",
133 "InvalidUrlRedirectClientError",
134 "NonHttpUrlRedirectClientError",
135 "ServerConnectionError",
136 "ServerDisconnectedError",
137 "ServerFingerprintMismatch",
138 "ServerTimeoutError",
139 "SocketTimeoutError",
140 "TooManyRedirects",
141 "WSServerHandshakeError",
142 # client_reqrep
143 "ClientRequest",
144 "ClientResponse",
145 "Fingerprint",
146 "RequestInfo",
147 # connector
148 "BaseConnector",
149 "TCPConnector",
150 "UnixConnector",
151 "NamedPipeConnector",
152 # client_ws
153 "ClientWebSocketResponse",
154 # client
155 "ClientSession",
156 "ClientTimeout",
157 "ClientWSTimeout",
158 "request",
159 "WSMessageTypeError",
160)
161
162
163if TYPE_CHECKING:
164 from ssl import SSLContext
165else:
166 SSLContext = None
167
168if sys.version_info >= (3, 11) and TYPE_CHECKING:
169 from typing import Unpack
170
171
172class _RequestOptions(TypedDict, total=False):
173 params: Query
174 data: Any
175 json: Any
176 cookies: Union[LooseCookies, None]
177 headers: Union[LooseHeaders, None]
178 skip_auto_headers: Union[Iterable[str], None]
179 auth: Union[BasicAuth, None]
180 allow_redirects: bool
181 max_redirects: int
182 compress: Union[str, bool, None]
183 chunked: Union[bool, None]
184 expect100: bool
185 raise_for_status: Union[None, bool, Callable[[ClientResponse], Awaitable[None]]]
186 read_until_eof: bool
187 proxy: Union[StrOrURL, None]
188 proxy_auth: Union[BasicAuth, None]
189 timeout: "Union[ClientTimeout, _SENTINEL, None]"
190 ssl: Union[SSLContext, bool, Fingerprint]
191 server_hostname: Union[str, None]
192 proxy_headers: Union[LooseHeaders, None]
193 trace_request_ctx: Union[Mapping[str, Any], None]
194 read_bufsize: Union[int, None]
195 auto_decompress: Union[bool, None]
196 max_line_size: Union[int, None]
197 max_field_size: Union[int, None]
198 max_headers: Union[int, None]
199 middlewares: Optional[Sequence[ClientMiddlewareType]]
200
201
202@attr.s(auto_attribs=True, frozen=True, slots=True)
203class ClientTimeout:
204 total: Optional[float] = None
205 connect: Optional[float] = None
206 sock_read: Optional[float] = None
207 sock_connect: Optional[float] = None
208 ceil_threshold: float = 5
209
210 # pool_queue_timeout: Optional[float] = None
211 # dns_resolution_timeout: Optional[float] = None
212 # socket_connect_timeout: Optional[float] = None
213 # connection_acquiring_timeout: Optional[float] = None
214 # new_connection_timeout: Optional[float] = None
215 # http_header_timeout: Optional[float] = None
216 # response_body_timeout: Optional[float] = None
217
218 # to create a timeout specific for a single request, either
219 # - create a completely new one to overwrite the default
220 # - or use http://www.attrs.org/en/stable/api.html#attr.evolve
221 # to overwrite the defaults
222
223
224# 5 Minute default read timeout
225DEFAULT_TIMEOUT: Final[ClientTimeout] = ClientTimeout(total=5 * 60, sock_connect=30)
226
227# https://www.rfc-editor.org/rfc/rfc9110#section-9.2.2
228IDEMPOTENT_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE", "PUT", "DELETE"})
229
230_RetType = TypeVar("_RetType", ClientResponse, ClientWebSocketResponse)
231_CharsetResolver = Callable[[ClientResponse, bytes], str]
232
233
234class ClientSession:
235 """First-class interface for making HTTP requests."""
236
237 ATTRS = frozenset(
238 [
239 "_base_url",
240 "_base_url_origin",
241 "_source_traceback",
242 "_connector",
243 "_loop",
244 "_cookie_jar",
245 "_connector_owner",
246 "_default_auth",
247 "_version",
248 "_json_serialize",
249 "_requote_redirect_url",
250 "_timeout",
251 "_raise_for_status",
252 "_auto_decompress",
253 "_trust_env",
254 "_default_headers",
255 "_skip_auto_headers",
256 "_request_class",
257 "_response_class",
258 "_ws_response_class",
259 "_trace_configs",
260 "_read_bufsize",
261 "_max_line_size",
262 "_max_field_size",
263 "_max_headers",
264 "_resolve_charset",
265 "_default_proxy",
266 "_default_proxy_auth",
267 "_retry_connection",
268 "_middlewares",
269 "requote_redirect_url",
270 ]
271 )
272
273 _source_traceback: Optional[traceback.StackSummary] = None
274 _connector: Optional[BaseConnector] = None
275
276 def __init__(
277 self,
278 base_url: Optional[StrOrURL] = None,
279 *,
280 connector: Optional[BaseConnector] = None,
281 loop: Optional[asyncio.AbstractEventLoop] = None,
282 cookies: Optional[LooseCookies] = None,
283 headers: Optional[LooseHeaders] = None,
284 proxy: Optional[StrOrURL] = None,
285 proxy_auth: Optional[BasicAuth] = None,
286 skip_auto_headers: Optional[Iterable[str]] = None,
287 auth: Optional[BasicAuth] = None,
288 json_serialize: JSONEncoder = json.dumps,
289 request_class: Type[ClientRequest] = ClientRequest,
290 response_class: Type[ClientResponse] = ClientResponse,
291 ws_response_class: Type[ClientWebSocketResponse] = ClientWebSocketResponse,
292 version: HttpVersion = http.HttpVersion11,
293 cookie_jar: Optional[AbstractCookieJar] = None,
294 connector_owner: bool = True,
295 raise_for_status: Union[
296 bool, Callable[[ClientResponse], Awaitable[None]]
297 ] = False,
298 read_timeout: Union[float, _SENTINEL] = sentinel,
299 conn_timeout: Optional[float] = None,
300 timeout: Union[object, ClientTimeout] = sentinel,
301 auto_decompress: bool = True,
302 trust_env: bool = False,
303 requote_redirect_url: bool = True,
304 trace_configs: Optional[List[TraceConfig]] = None,
305 read_bufsize: int = 2**16,
306 max_line_size: int = 8190,
307 max_field_size: int = 8190,
308 max_headers: int = 128,
309 fallback_charset_resolver: _CharsetResolver = lambda r, b: "utf-8",
310 middlewares: Sequence[ClientMiddlewareType] = (),
311 ssl_shutdown_timeout: Union[_SENTINEL, None, float] = sentinel,
312 ) -> None:
313 # We initialise _connector to None immediately, as it's referenced in __del__()
314 # and could cause issues if an exception occurs during initialisation.
315 self._connector: Optional[BaseConnector] = None
316
317 if loop is None:
318 if connector is not None:
319 loop = connector._loop
320
321 loop = loop or asyncio.get_running_loop()
322
323 if base_url is None or isinstance(base_url, URL):
324 self._base_url: Optional[URL] = base_url
325 self._base_url_origin = None if base_url is None else base_url.origin()
326 else:
327 self._base_url = URL(base_url)
328 self._base_url_origin = self._base_url.origin()
329 assert self._base_url.absolute, "Only absolute URLs are supported"
330 if self._base_url is not None and not self._base_url.path.endswith("/"):
331 raise ValueError("base_url must have a trailing '/'")
332
333 if timeout is sentinel or timeout is None:
334 self._timeout = DEFAULT_TIMEOUT
335 if read_timeout is not sentinel:
336 warnings.warn(
337 "read_timeout is deprecated, use timeout argument instead",
338 DeprecationWarning,
339 stacklevel=2,
340 )
341 self._timeout = attr.evolve(self._timeout, total=read_timeout)
342 if conn_timeout is not None:
343 self._timeout = attr.evolve(self._timeout, connect=conn_timeout)
344 warnings.warn(
345 "conn_timeout is deprecated, use timeout argument instead",
346 DeprecationWarning,
347 stacklevel=2,
348 )
349 else:
350 if not isinstance(timeout, ClientTimeout):
351 raise ValueError(
352 f"timeout parameter cannot be of {type(timeout)} type, "
353 "please use 'timeout=ClientTimeout(...)'",
354 )
355 self._timeout = timeout
356 if read_timeout is not sentinel:
357 raise ValueError(
358 "read_timeout and timeout parameters "
359 "conflict, please setup "
360 "timeout.read"
361 )
362 if conn_timeout is not None:
363 raise ValueError(
364 "conn_timeout and timeout parameters "
365 "conflict, please setup "
366 "timeout.connect"
367 )
368
369 if ssl_shutdown_timeout is not sentinel:
370 warnings.warn(
371 "The ssl_shutdown_timeout parameter is deprecated and will be removed in aiohttp 4.0",
372 DeprecationWarning,
373 stacklevel=2,
374 )
375
376 if connector is None:
377 connector = TCPConnector(
378 loop=loop, ssl_shutdown_timeout=ssl_shutdown_timeout
379 )
380
381 if connector._loop is not loop:
382 raise RuntimeError("Session and connector has to use same event loop")
383
384 self._loop = loop
385
386 if loop.get_debug():
387 self._source_traceback = traceback.extract_stack(sys._getframe(1))
388
389 if cookie_jar is None:
390 cookie_jar = CookieJar(loop=loop)
391 self._cookie_jar = cookie_jar
392
393 if cookies:
394 self._cookie_jar.update_cookies(cookies)
395
396 self._connector = connector
397 self._connector_owner = connector_owner
398 self._default_auth = auth
399 self._version = version
400 self._json_serialize = json_serialize
401 self._raise_for_status = raise_for_status
402 self._auto_decompress = auto_decompress
403 self._trust_env = trust_env
404 self._requote_redirect_url = requote_redirect_url
405 self._read_bufsize = read_bufsize
406 self._max_line_size = max_line_size
407 self._max_field_size = max_field_size
408 self._max_headers = max_headers
409
410 # Convert to list of tuples
411 if headers:
412 real_headers: CIMultiDict[str] = CIMultiDict(headers)
413 else:
414 real_headers = CIMultiDict()
415 self._default_headers: CIMultiDict[str] = real_headers
416 if skip_auto_headers is not None:
417 self._skip_auto_headers = frozenset(istr(i) for i in skip_auto_headers)
418 else:
419 self._skip_auto_headers = frozenset()
420
421 self._request_class = request_class
422 self._response_class = response_class
423 self._ws_response_class = ws_response_class
424
425 self._trace_configs = trace_configs or []
426 for trace_config in self._trace_configs:
427 trace_config.freeze()
428
429 self._resolve_charset = fallback_charset_resolver
430
431 self._default_proxy = proxy
432 self._default_proxy_auth = proxy_auth
433 self._retry_connection: bool = True
434 self._middlewares = middlewares
435
436 def __init_subclass__(cls: Type["ClientSession"]) -> None:
437 warnings.warn(
438 "Inheritance class {} from ClientSession "
439 "is discouraged".format(cls.__name__),
440 DeprecationWarning,
441 stacklevel=2,
442 )
443
444 if DEBUG:
445
446 def __setattr__(self, name: str, val: Any) -> None:
447 if name not in self.ATTRS:
448 warnings.warn(
449 "Setting custom ClientSession.{} attribute "
450 "is discouraged".format(name),
451 DeprecationWarning,
452 stacklevel=2,
453 )
454 super().__setattr__(name, val)
455
456 def __del__(self, _warnings: Any = warnings) -> None:
457 if not self.closed:
458 kwargs = {"source": self}
459 _warnings.warn(
460 f"Unclosed client session {self!r}", ResourceWarning, **kwargs
461 )
462 context = {"client_session": self, "message": "Unclosed client session"}
463 if self._source_traceback is not None:
464 context["source_traceback"] = self._source_traceback
465 self._loop.call_exception_handler(context)
466
467 if sys.version_info >= (3, 11) and TYPE_CHECKING:
468
469 def request(
470 self,
471 method: str,
472 url: StrOrURL,
473 **kwargs: Unpack[_RequestOptions],
474 ) -> "_RequestContextManager": ...
475
476 else:
477
478 def request(
479 self, method: str, url: StrOrURL, **kwargs: Any
480 ) -> "_RequestContextManager":
481 """Perform HTTP request."""
482 return _RequestContextManager(self._request(method, url, **kwargs))
483
484 def _build_url(self, str_or_url: StrOrURL) -> URL:
485 url = URL(str_or_url)
486 if self._base_url and not url.absolute:
487 return self._base_url.join(url)
488 return url
489
490 async def _request(
491 self,
492 method: str,
493 str_or_url: StrOrURL,
494 *,
495 params: Query = None,
496 data: Any = None,
497 json: Any = None,
498 cookies: Optional[LooseCookies] = None,
499 headers: Optional[LooseHeaders] = None,
500 skip_auto_headers: Optional[Iterable[str]] = None,
501 auth: Optional[BasicAuth] = None,
502 allow_redirects: bool = True,
503 max_redirects: int = 10,
504 compress: Union[str, bool, None] = None,
505 chunked: Optional[bool] = None,
506 expect100: bool = False,
507 raise_for_status: Union[
508 None, bool, Callable[[ClientResponse], Awaitable[None]]
509 ] = None,
510 read_until_eof: bool = True,
511 proxy: Optional[StrOrURL] = None,
512 proxy_auth: Optional[BasicAuth] = None,
513 timeout: Union[ClientTimeout, _SENTINEL] = sentinel,
514 verify_ssl: Optional[bool] = None,
515 fingerprint: Optional[bytes] = None,
516 ssl_context: Optional[SSLContext] = None,
517 ssl: Union[SSLContext, bool, Fingerprint] = True,
518 server_hostname: Optional[str] = None,
519 proxy_headers: Optional[LooseHeaders] = None,
520 trace_request_ctx: Optional[Mapping[str, Any]] = None,
521 read_bufsize: Optional[int] = None,
522 auto_decompress: Optional[bool] = None,
523 max_line_size: Optional[int] = None,
524 max_field_size: Optional[int] = None,
525 max_headers: Optional[int] = None,
526 middlewares: Optional[Sequence[ClientMiddlewareType]] = None,
527 ) -> ClientResponse:
528
529 # NOTE: timeout clamps existing connect and read timeouts. We cannot
530 # set the default to None because we need to detect if the user wants
531 # to use the existing timeouts by setting timeout to None.
532
533 if self.closed:
534 raise RuntimeError("Session is closed")
535
536 ssl = _merge_ssl_params(ssl, verify_ssl, ssl_context, fingerprint)
537
538 if data is not None and json is not None:
539 raise ValueError(
540 "data and json parameters can not be used at the same time"
541 )
542 elif json is not None:
543 data = payload.JsonPayload(json, dumps=self._json_serialize)
544
545 if not isinstance(chunked, bool) and chunked is not None:
546 warnings.warn("Chunk size is deprecated #1615", DeprecationWarning)
547
548 redirects = 0
549 history: List[ClientResponse] = []
550 version = self._version
551 params = params or {}
552
553 # Merge with default headers and transform to CIMultiDict
554 headers = self._prepare_headers(headers)
555
556 try:
557 url = self._build_url(str_or_url)
558 except ValueError as e:
559 raise InvalidUrlClientError(str_or_url) from e
560
561 assert self._connector is not None
562 if url.scheme not in self._connector.allowed_protocol_schema_set:
563 raise NonHttpUrlClientError(url)
564
565 skip_headers: Optional[Iterable[istr]]
566 if skip_auto_headers is not None:
567 skip_headers = {
568 istr(i) for i in skip_auto_headers
569 } | self._skip_auto_headers
570 elif self._skip_auto_headers:
571 skip_headers = self._skip_auto_headers
572 else:
573 skip_headers = None
574
575 if proxy is None:
576 proxy = self._default_proxy
577 if proxy_auth is None:
578 proxy_auth = self._default_proxy_auth
579
580 if proxy is None:
581 proxy_headers = None
582 else:
583 proxy_headers = self._prepare_headers(proxy_headers)
584 try:
585 proxy = URL(proxy)
586 except ValueError as e:
587 raise InvalidURL(proxy) from e
588
589 if timeout is sentinel:
590 real_timeout: ClientTimeout = self._timeout
591 else:
592 if not isinstance(timeout, ClientTimeout):
593 real_timeout = ClientTimeout(total=timeout)
594 else:
595 real_timeout = timeout
596 # timeout is cumulative for all request operations
597 # (request, redirects, responses, data consuming)
598 tm = TimeoutHandle(
599 self._loop, real_timeout.total, ceil_threshold=real_timeout.ceil_threshold
600 )
601 handle = tm.start()
602
603 if read_bufsize is None:
604 read_bufsize = self._read_bufsize
605
606 if auto_decompress is None:
607 auto_decompress = self._auto_decompress
608
609 if max_line_size is None:
610 max_line_size = self._max_line_size
611
612 if max_field_size is None:
613 max_field_size = self._max_field_size
614
615 if max_headers is None:
616 max_headers = self._max_headers
617
618 traces = [
619 Trace(
620 self,
621 trace_config,
622 trace_config.trace_config_ctx(trace_request_ctx=trace_request_ctx),
623 )
624 for trace_config in self._trace_configs
625 ]
626
627 for trace in traces:
628 await trace.send_request_start(method, url.update_query(params), headers)
629
630 timer = tm.timer()
631 try:
632 with timer:
633 # https://www.rfc-editor.org/rfc/rfc9112.html#name-retrying-requests
634 retry_persistent_connection = (
635 self._retry_connection and method in IDEMPOTENT_METHODS
636 )
637 while True:
638 url, auth_from_url = strip_auth_from_url(url)
639 if not url.raw_host:
640 # NOTE: Bail early, otherwise, causes `InvalidURL` through
641 # NOTE: `self._request_class()` below.
642 err_exc_cls = (
643 InvalidUrlRedirectClientError
644 if redirects
645 else InvalidUrlClientError
646 )
647 raise err_exc_cls(url)
648 # If `auth` was passed for an already authenticated URL,
649 # disallow only if this is the initial URL; this is to avoid issues
650 # with sketchy redirects that are not the caller's responsibility
651 if not history and (auth and auth_from_url):
652 raise ValueError(
653 "Cannot combine AUTH argument with "
654 "credentials encoded in URL"
655 )
656
657 # Override the auth with the one from the URL only if we
658 # have no auth, or if we got an auth from a redirect URL
659 if auth is None or (history and auth_from_url is not None):
660 auth = auth_from_url
661
662 if (
663 auth is None
664 and self._default_auth
665 and (
666 not self._base_url or self._base_url_origin == url.origin()
667 )
668 ):
669 auth = self._default_auth
670
671 # Try netrc if auth is still None and trust_env is enabled.
672 if auth is None and self._trust_env and url.host is not None:
673 auth = await self._loop.run_in_executor(
674 None, self._get_netrc_auth, url.host
675 )
676
677 # It would be confusing if we support explicit
678 # Authorization header with auth argument
679 if (
680 headers is not None
681 and auth is not None
682 and hdrs.AUTHORIZATION in headers
683 ):
684 raise ValueError(
685 "Cannot combine AUTHORIZATION header "
686 "with AUTH argument or credentials "
687 "encoded in URL"
688 )
689
690 all_cookies = self._cookie_jar.filter_cookies(url)
691
692 if cookies is not None:
693 tmp_cookie_jar = CookieJar(
694 quote_cookie=self._cookie_jar.quote_cookie
695 )
696 tmp_cookie_jar.update_cookies(cookies)
697 req_cookies = tmp_cookie_jar.filter_cookies(url)
698 if req_cookies:
699 all_cookies.load(req_cookies)
700
701 proxy_: Optional[URL] = None
702 if proxy is not None:
703 proxy_ = URL(proxy)
704 elif self._trust_env:
705 with suppress(LookupError):
706 proxy_, proxy_auth = await asyncio.to_thread(
707 get_env_proxy_for_url, url
708 )
709
710 req = self._request_class(
711 method,
712 url,
713 params=params,
714 headers=headers,
715 skip_auto_headers=skip_headers,
716 data=data,
717 cookies=all_cookies,
718 auth=auth,
719 version=version,
720 compress=compress,
721 chunked=chunked,
722 expect100=expect100,
723 loop=self._loop,
724 response_class=self._response_class,
725 proxy=proxy_,
726 proxy_auth=proxy_auth,
727 timer=timer,
728 session=self,
729 ssl=ssl if ssl is not None else True,
730 server_hostname=server_hostname,
731 proxy_headers=proxy_headers,
732 traces=traces,
733 trust_env=self.trust_env,
734 )
735
736 async def _connect_and_send_request(
737 req: ClientRequest,
738 ) -> ClientResponse:
739 # connection timeout
740 assert self._connector is not None
741 try:
742 conn = await self._connector.connect(
743 req, traces=traces, timeout=real_timeout
744 )
745 except asyncio.TimeoutError as exc:
746 raise ConnectionTimeoutError(
747 f"Connection timeout to host {req.url}"
748 ) from exc
749
750 assert conn.protocol is not None
751 conn.protocol.set_response_params(
752 timer=timer,
753 skip_payload=req.method in EMPTY_BODY_METHODS,
754 read_until_eof=read_until_eof,
755 auto_decompress=auto_decompress,
756 read_timeout=real_timeout.sock_read,
757 read_bufsize=read_bufsize,
758 timeout_ceil_threshold=self._connector._timeout_ceil_threshold,
759 max_line_size=max_line_size,
760 max_field_size=max_field_size,
761 max_headers=max_headers,
762 )
763 try:
764 resp = await req.send(conn)
765 try:
766 await resp.start(conn)
767 except BaseException:
768 resp.close()
769 raise
770 except BaseException:
771 conn.close()
772 raise
773 return resp
774
775 # Apply middleware (if any) - per-request middleware overrides session middleware
776 effective_middlewares = (
777 self._middlewares if middlewares is None else middlewares
778 )
779
780 if effective_middlewares:
781 handler = build_client_middlewares(
782 _connect_and_send_request, effective_middlewares
783 )
784 else:
785 handler = _connect_and_send_request
786
787 try:
788 resp = await handler(req)
789 # Client connector errors should not be retried
790 except (
791 ConnectionTimeoutError,
792 ClientConnectorError,
793 ClientConnectorCertificateError,
794 ClientConnectorSSLError,
795 ):
796 raise
797 except (ClientOSError, ServerDisconnectedError):
798 if retry_persistent_connection:
799 retry_persistent_connection = False
800 continue
801 raise
802 except ClientError:
803 raise
804 except OSError as exc:
805 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
806 raise
807 raise ClientOSError(*exc.args) from exc
808
809 # Update cookies from raw headers to preserve duplicates
810 if resp._raw_cookie_headers:
811 self._cookie_jar.update_cookies_from_headers(
812 resp._raw_cookie_headers, resp.url
813 )
814
815 # redirects
816 if resp.status in (301, 302, 303, 307, 308) and allow_redirects:
817
818 for trace in traces:
819 await trace.send_request_redirect(
820 method, url.update_query(params), headers, resp
821 )
822
823 redirects += 1
824 history.append(resp)
825 if max_redirects and redirects >= max_redirects:
826 if req._body is not None:
827 await req._body.close()
828 resp.close()
829 raise TooManyRedirects(
830 history[0].request_info, tuple(history)
831 )
832
833 # For 301 and 302, mimic IE, now changed in RFC
834 # https://github.com/kennethreitz/requests/pull/269
835 if (resp.status == 303 and resp.method != hdrs.METH_HEAD) or (
836 resp.status in (301, 302) and resp.method == hdrs.METH_POST
837 ):
838 method = hdrs.METH_GET
839 data = None
840 if headers.get(hdrs.CONTENT_LENGTH):
841 headers.pop(hdrs.CONTENT_LENGTH)
842 else:
843 # For 307/308, always preserve the request body
844 # For 301/302 with non-POST methods, preserve the request body
845 # https://www.rfc-editor.org/rfc/rfc9110#section-15.4.3-3.1
846 # Use the existing payload to avoid recreating it from a potentially consumed file
847 data = req._body
848
849 r_url = resp.headers.get(hdrs.LOCATION) or resp.headers.get(
850 hdrs.URI
851 )
852 if r_url is None:
853 # see github.com/aio-libs/aiohttp/issues/2022
854 break
855 else:
856 # reading from correct redirection
857 # response is forbidden
858 resp.release()
859
860 try:
861 parsed_redirect_url = URL(
862 r_url, encoded=not self._requote_redirect_url
863 )
864 except ValueError as e:
865 if req._body is not None:
866 await req._body.close()
867 resp.close()
868 raise InvalidUrlRedirectClientError(
869 r_url,
870 "Server attempted redirecting to a location that does not look like a URL",
871 ) from e
872
873 scheme = parsed_redirect_url.scheme
874 if scheme not in HTTP_AND_EMPTY_SCHEMA_SET:
875 if req._body is not None:
876 await req._body.close()
877 resp.close()
878 raise NonHttpUrlRedirectClientError(r_url)
879 elif not scheme:
880 parsed_redirect_url = url.join(parsed_redirect_url)
881
882 try:
883 redirect_origin = parsed_redirect_url.origin()
884 except ValueError as origin_val_err:
885 if req._body is not None:
886 await req._body.close()
887 resp.close()
888 raise InvalidUrlRedirectClientError(
889 parsed_redirect_url,
890 "Invalid redirect URL origin",
891 ) from origin_val_err
892
893 if url.origin() != redirect_origin:
894 auth = None
895 headers.pop(hdrs.AUTHORIZATION, None)
896 headers.pop(hdrs.COOKIE, None)
897 headers.pop(hdrs.PROXY_AUTHORIZATION, None)
898
899 url = parsed_redirect_url
900 params = {}
901 resp.release()
902 continue
903
904 break
905
906 if req._body is not None:
907 await req._body.close()
908 # check response status
909 if raise_for_status is None:
910 raise_for_status = self._raise_for_status
911
912 if raise_for_status is None:
913 pass
914 elif callable(raise_for_status):
915 await raise_for_status(resp)
916 elif raise_for_status:
917 resp.raise_for_status()
918
919 # register connection
920 if handle is not None:
921 if resp.connection is not None:
922 resp.connection.add_callback(handle.cancel)
923 else:
924 handle.cancel()
925
926 resp._history = tuple(history)
927
928 for trace in traces:
929 await trace.send_request_end(
930 method, url.update_query(params), headers, resp
931 )
932 return resp
933
934 except BaseException as e:
935 # cleanup timer
936 tm.close()
937 if handle:
938 handle.cancel()
939 handle = None
940
941 for trace in traces:
942 await trace.send_request_exception(
943 method, url.update_query(params), headers, e
944 )
945 raise
946
947 def ws_connect(
948 self,
949 url: StrOrURL,
950 *,
951 method: str = hdrs.METH_GET,
952 protocols: Iterable[str] = (),
953 timeout: Union[ClientWSTimeout, _SENTINEL] = sentinel,
954 receive_timeout: Optional[float] = None,
955 autoclose: bool = True,
956 autoping: bool = True,
957 heartbeat: Optional[float] = None,
958 auth: Optional[BasicAuth] = None,
959 origin: Optional[str] = None,
960 params: Query = None,
961 headers: Optional[LooseHeaders] = None,
962 proxy: Optional[StrOrURL] = None,
963 proxy_auth: Optional[BasicAuth] = None,
964 ssl: Union[SSLContext, bool, Fingerprint] = True,
965 verify_ssl: Optional[bool] = None,
966 fingerprint: Optional[bytes] = None,
967 ssl_context: Optional[SSLContext] = None,
968 server_hostname: Optional[str] = None,
969 proxy_headers: Optional[LooseHeaders] = None,
970 compress: int = 0,
971 max_msg_size: int = 4 * 1024 * 1024,
972 ) -> "_WSRequestContextManager":
973 """Initiate websocket connection."""
974 return _WSRequestContextManager(
975 self._ws_connect(
976 url,
977 method=method,
978 protocols=protocols,
979 timeout=timeout,
980 receive_timeout=receive_timeout,
981 autoclose=autoclose,
982 autoping=autoping,
983 heartbeat=heartbeat,
984 auth=auth,
985 origin=origin,
986 params=params,
987 headers=headers,
988 proxy=proxy,
989 proxy_auth=proxy_auth,
990 ssl=ssl,
991 verify_ssl=verify_ssl,
992 fingerprint=fingerprint,
993 ssl_context=ssl_context,
994 server_hostname=server_hostname,
995 proxy_headers=proxy_headers,
996 compress=compress,
997 max_msg_size=max_msg_size,
998 )
999 )
1000
1001 async def _ws_connect(
1002 self,
1003 url: StrOrURL,
1004 *,
1005 method: str = hdrs.METH_GET,
1006 protocols: Iterable[str] = (),
1007 timeout: Union[ClientWSTimeout, _SENTINEL] = sentinel,
1008 receive_timeout: Optional[float] = None,
1009 autoclose: bool = True,
1010 autoping: bool = True,
1011 heartbeat: Optional[float] = None,
1012 auth: Optional[BasicAuth] = None,
1013 origin: Optional[str] = None,
1014 params: Query = None,
1015 headers: Optional[LooseHeaders] = None,
1016 proxy: Optional[StrOrURL] = None,
1017 proxy_auth: Optional[BasicAuth] = None,
1018 ssl: Union[SSLContext, bool, Fingerprint] = True,
1019 verify_ssl: Optional[bool] = None,
1020 fingerprint: Optional[bytes] = None,
1021 ssl_context: Optional[SSLContext] = None,
1022 server_hostname: Optional[str] = None,
1023 proxy_headers: Optional[LooseHeaders] = None,
1024 compress: int = 0,
1025 max_msg_size: int = 4 * 1024 * 1024,
1026 ) -> ClientWebSocketResponse:
1027 if timeout is not sentinel:
1028 if isinstance(timeout, ClientWSTimeout):
1029 ws_timeout = timeout
1030 else:
1031 warnings.warn(
1032 "parameter 'timeout' of type 'float' "
1033 "is deprecated, please use "
1034 "'timeout=ClientWSTimeout(ws_close=...)'",
1035 DeprecationWarning,
1036 stacklevel=2,
1037 )
1038 ws_timeout = ClientWSTimeout(ws_close=timeout)
1039 else:
1040 ws_timeout = DEFAULT_WS_CLIENT_TIMEOUT
1041 if receive_timeout is not None:
1042 warnings.warn(
1043 "float parameter 'receive_timeout' "
1044 "is deprecated, please use parameter "
1045 "'timeout=ClientWSTimeout(ws_receive=...)'",
1046 DeprecationWarning,
1047 stacklevel=2,
1048 )
1049 ws_timeout = attr.evolve(ws_timeout, ws_receive=receive_timeout)
1050
1051 if headers is None:
1052 real_headers: CIMultiDict[str] = CIMultiDict()
1053 else:
1054 real_headers = CIMultiDict(headers)
1055
1056 default_headers = {
1057 hdrs.UPGRADE: "websocket",
1058 hdrs.CONNECTION: "Upgrade",
1059 hdrs.SEC_WEBSOCKET_VERSION: "13",
1060 }
1061
1062 for key, value in default_headers.items():
1063 real_headers.setdefault(key, value)
1064
1065 sec_key = base64.b64encode(os.urandom(16))
1066 real_headers[hdrs.SEC_WEBSOCKET_KEY] = sec_key.decode()
1067
1068 if protocols:
1069 real_headers[hdrs.SEC_WEBSOCKET_PROTOCOL] = ",".join(protocols)
1070 if origin is not None:
1071 real_headers[hdrs.ORIGIN] = origin
1072 if compress:
1073 extstr = ws_ext_gen(compress=compress)
1074 real_headers[hdrs.SEC_WEBSOCKET_EXTENSIONS] = extstr
1075
1076 # For the sake of backward compatibility, if user passes in None, convert it to True
1077 if ssl is None:
1078 warnings.warn(
1079 "ssl=None is deprecated, please use ssl=True",
1080 DeprecationWarning,
1081 stacklevel=2,
1082 )
1083 ssl = True
1084 ssl = _merge_ssl_params(ssl, verify_ssl, ssl_context, fingerprint)
1085
1086 # send request
1087 resp = await self.request(
1088 method,
1089 url,
1090 params=params,
1091 headers=real_headers,
1092 read_until_eof=False,
1093 auth=auth,
1094 proxy=proxy,
1095 proxy_auth=proxy_auth,
1096 ssl=ssl,
1097 server_hostname=server_hostname,
1098 proxy_headers=proxy_headers,
1099 )
1100
1101 try:
1102 # check handshake
1103 if resp.status != 101:
1104 raise WSServerHandshakeError(
1105 resp.request_info,
1106 resp.history,
1107 message="Invalid response status",
1108 status=resp.status,
1109 headers=resp.headers,
1110 )
1111
1112 if resp.headers.get(hdrs.UPGRADE, "").lower() != "websocket":
1113 raise WSServerHandshakeError(
1114 resp.request_info,
1115 resp.history,
1116 message="Invalid upgrade header",
1117 status=resp.status,
1118 headers=resp.headers,
1119 )
1120
1121 if resp.headers.get(hdrs.CONNECTION, "").lower() != "upgrade":
1122 raise WSServerHandshakeError(
1123 resp.request_info,
1124 resp.history,
1125 message="Invalid connection header",
1126 status=resp.status,
1127 headers=resp.headers,
1128 )
1129
1130 # key calculation
1131 r_key = resp.headers.get(hdrs.SEC_WEBSOCKET_ACCEPT, "")
1132 match = base64.b64encode(hashlib.sha1(sec_key + WS_KEY).digest()).decode()
1133 if r_key != match:
1134 raise WSServerHandshakeError(
1135 resp.request_info,
1136 resp.history,
1137 message="Invalid challenge response",
1138 status=resp.status,
1139 headers=resp.headers,
1140 )
1141
1142 # websocket protocol
1143 protocol = None
1144 if protocols and hdrs.SEC_WEBSOCKET_PROTOCOL in resp.headers:
1145 resp_protocols = [
1146 proto.strip()
1147 for proto in resp.headers[hdrs.SEC_WEBSOCKET_PROTOCOL].split(",")
1148 ]
1149
1150 for proto in resp_protocols:
1151 if proto in protocols:
1152 protocol = proto
1153 break
1154
1155 # websocket compress
1156 notakeover = False
1157 if compress:
1158 compress_hdrs = resp.headers.get(hdrs.SEC_WEBSOCKET_EXTENSIONS)
1159 if compress_hdrs:
1160 try:
1161 compress, notakeover = ws_ext_parse(compress_hdrs)
1162 except WSHandshakeError as exc:
1163 raise WSServerHandshakeError(
1164 resp.request_info,
1165 resp.history,
1166 message=exc.args[0],
1167 status=resp.status,
1168 headers=resp.headers,
1169 ) from exc
1170 else:
1171 compress = 0
1172 notakeover = False
1173
1174 conn = resp.connection
1175 assert conn is not None
1176 conn_proto = conn.protocol
1177 assert conn_proto is not None
1178
1179 # For WS connection the read_timeout must be either receive_timeout or greater
1180 # None == no timeout, i.e. infinite timeout, so None is the max timeout possible
1181 if ws_timeout.ws_receive is None:
1182 # Reset regardless
1183 conn_proto.read_timeout = None
1184 elif conn_proto.read_timeout is not None:
1185 conn_proto.read_timeout = max(
1186 ws_timeout.ws_receive, conn_proto.read_timeout
1187 )
1188
1189 transport = conn.transport
1190 assert transport is not None
1191 reader = WebSocketDataQueue(conn_proto, 2**16, loop=self._loop)
1192 conn_proto.set_parser(WebSocketReader(reader, max_msg_size), reader)
1193 writer = WebSocketWriter(
1194 conn_proto,
1195 transport,
1196 use_mask=True,
1197 compress=compress,
1198 notakeover=notakeover,
1199 )
1200 except BaseException: