Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiohttp/connector.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 functools
3import random
4import socket
5import sys
6import traceback
7import warnings
8from asyncio.base_events import BaseEventLoop
9from collections import OrderedDict, defaultdict, deque
10from collections.abc import Awaitable, Callable, Iterator, Sequence
11from contextlib import suppress
12from http import HTTPStatus
13from itertools import chain, cycle, islice
14from time import monotonic
15from types import TracebackType
16from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast
18import aiohappyeyeballs
19from aiohappyeyeballs import AddrInfoType, SocketFactoryType
21from . import hdrs, helpers
22from .abc import AbstractResolver, ResolveResult
23from .client_exceptions import (
24 ClientConnectionError,
25 ClientConnectorCertificateError,
26 ClientConnectorDNSError,
27 ClientConnectorError,
28 ClientConnectorSSLError,
29 ClientHttpProxyError,
30 ClientProxyConnectionError,
31 InvalidUrlClientError,
32 ServerFingerprintMismatch,
33 UnixClientConnectorError,
34 cert_errors,
35 ssl_errors,
36)
37from .client_proto import ResponseHandler
38from .client_reqrep import ClientRequest, Fingerprint, _merge_ssl_params
39from .helpers import (
40 _SENTINEL,
41 ceil_timeout,
42 is_canonical_ipv4_address,
43 is_ip_address,
44 noop,
45 sentinel,
46 set_exception,
47 set_result,
48)
49from .log import client_logger
50from .resolver import DefaultResolver
52if sys.version_info >= (3, 12):
53 from collections.abc import Buffer
54else:
55 Buffer = Union[bytes, bytearray, "memoryview[int]", "memoryview[bytes]"]
57if TYPE_CHECKING:
58 import ssl
60 SSLContext = ssl.SSLContext
61else:
62 try:
63 import ssl
65 SSLContext = ssl.SSLContext
66 except ImportError: # pragma: no cover
67 ssl = None # type: ignore[assignment]
68 SSLContext = object # type: ignore[misc,assignment]
70EMPTY_SCHEMA_SET = frozenset({""})
71HTTP_SCHEMA_SET = frozenset({"http", "https"})
72WS_SCHEMA_SET = frozenset({"ws", "wss"})
74HTTP_AND_EMPTY_SCHEMA_SET = HTTP_SCHEMA_SET | EMPTY_SCHEMA_SET
75HIGH_LEVEL_SCHEMA_SET = HTTP_AND_EMPTY_SCHEMA_SET | WS_SCHEMA_SET
77NEEDS_CLEANUP_CLOSED = (3, 13, 0) <= sys.version_info < (
78 3,
79 13,
80 1,
81) or sys.version_info < (3, 12, 8)
82# Cleanup closed is no longer needed after https://github.com/python/cpython/pull/118960
83# which first appeared in Python 3.12.8 and 3.13.1
86__all__ = (
87 "BaseConnector",
88 "TCPConnector",
89 "UnixConnector",
90 "NamedPipeConnector",
91 "AddrInfoType",
92 "SocketFactoryType",
93)
96if TYPE_CHECKING:
97 from .client import ClientTimeout
98 from .client_reqrep import ConnectionKey
99 from .tracing import Trace
102class _DeprecationWaiter:
103 __slots__ = ("_awaitable", "_awaited")
105 def __init__(self, awaitable: Awaitable[Any]) -> None:
106 self._awaitable = awaitable
107 self._awaited = False
109 def __await__(self) -> Any:
110 self._awaited = True
111 return self._awaitable.__await__()
113 def __del__(self) -> None:
114 if not self._awaited:
115 warnings.warn(
116 "Connector.close() is a coroutine, "
117 "please use await connector.close()",
118 DeprecationWarning,
119 )
122async def _wait_for_close(waiters: list[Awaitable[object]]) -> None:
123 """Wait for all waiters to finish closing."""
124 results = await asyncio.gather(*waiters, return_exceptions=True)
125 for res in results:
126 if isinstance(res, Exception):
127 client_logger.debug("Error while closing connector: %r", res)
130class Connection:
132 _source_traceback = None
134 def __init__(
135 self,
136 connector: "BaseConnector",
137 key: "ConnectionKey",
138 protocol: ResponseHandler,
139 loop: asyncio.AbstractEventLoop,
140 ) -> None:
141 self._key = key
142 self._connector = connector
143 self._loop = loop
144 self._protocol: ResponseHandler | None = protocol
145 self._callbacks: list[Callable[[], None]] = []
147 if loop.get_debug():
148 self._source_traceback = traceback.extract_stack(sys._getframe(1))
150 def __repr__(self) -> str:
151 return f"Connection<{self._key}>"
153 def __del__(self, _warnings: Any = warnings) -> None:
154 if self._protocol is not None:
155 kwargs = {"source": self}
156 _warnings.warn(f"Unclosed connection {self!r}", ResourceWarning, **kwargs)
157 if self._loop.is_closed():
158 return
160 self._connector._release(self._key, self._protocol, should_close=True)
162 context = {"client_connection": self, "message": "Unclosed connection"}
163 if self._source_traceback is not None:
164 context["source_traceback"] = self._source_traceback
165 self._loop.call_exception_handler(context)
167 def __bool__(self) -> Literal[True]:
168 """Force subclasses to not be falsy, to make checks simpler."""
169 return True
171 @property
172 def loop(self) -> asyncio.AbstractEventLoop:
173 warnings.warn(
174 "connector.loop property is deprecated", DeprecationWarning, stacklevel=2
175 )
176 return self._loop
178 @property
179 def transport(self) -> asyncio.Transport | None:
180 if self._protocol is None:
181 return None
182 return self._protocol.transport
184 @property
185 def protocol(self) -> ResponseHandler | None:
186 return self._protocol
188 def add_callback(self, callback: Callable[[], None]) -> None:
189 if callback is not None:
190 self._callbacks.append(callback)
192 def _notify_release(self) -> None:
193 callbacks, self._callbacks = self._callbacks[:], []
195 for cb in callbacks:
196 with suppress(Exception):
197 cb()
199 def close(self) -> None:
200 self._notify_release()
202 if self._protocol is not None:
203 self._connector._release(self._key, self._protocol, should_close=True)
204 self._protocol = None
206 def release(self) -> None:
207 self._notify_release()
209 if self._protocol is not None:
210 self._connector._release(self._key, self._protocol)
211 self._protocol = None
213 @property
214 def closed(self) -> bool:
215 return self._protocol is None or not self._protocol.is_connected()
218class _ConnectTunnelConnection(Connection):
219 """Special connection wrapper for CONNECT tunnels that must never be pooled.
221 This connection wraps the proxy connection that will be upgraded with TLS.
222 It must never be released to the pool because:
223 1. Its 'closed' future will never complete, causing session.close() to hang
224 2. It represents an intermediate state, not a reusable connection
225 3. The real connection (with TLS) will be created separately
226 """
228 def release(self) -> None:
229 """Do nothing - don't pool or close the connection.
231 These connections are an intermediate state during the CONNECT tunnel
232 setup and will be cleaned up naturally after the TLS upgrade. If they
233 were to be pooled, they would never be properly closed, causing
234 session.close() to wait forever for their 'closed' future.
235 """
238class _TransportPlaceholder:
239 """placeholder for BaseConnector.connect function"""
241 __slots__ = ("closed", "transport")
243 def __init__(self, closed_future: asyncio.Future[Exception | None]) -> None:
244 """Initialize a placeholder for a transport."""
245 self.closed = closed_future
246 self.transport = None
248 def close(self) -> None:
249 """Close the placeholder."""
251 def abort(self) -> None:
252 """Abort the placeholder (does nothing)."""
255class BaseConnector:
256 """Base connector class.
258 keepalive_timeout - (optional) Keep-alive timeout. None means a
259 reusable connection never expires from idling alone (a
260 disconnected one is discarded the next time the pool is
261 consulted for that host, not proactively). This is different
262 from force_close=True, which disables reuse entirely.
263 force_close - Set to True to force close and do reconnect
264 after each request (and between redirects).
265 limit - The total number of simultaneous connections.
266 limit_per_host - Number of simultaneous connections to one host.
267 enable_cleanup_closed - Enables clean-up closed ssl transports.
268 Disabled by default.
269 timeout_ceil_threshold - Trigger ceiling of timeout values when
270 it's above timeout_ceil_threshold.
271 loop - Optional event loop.
272 """
274 _closed = True # prevent AttributeError in __del__ if ctor was failed
275 _source_traceback = None
277 # abort transport after 2 seconds (cleanup broken connections)
278 _cleanup_closed_period = 2.0
280 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET
282 def __init__(
283 self,
284 *,
285 keepalive_timeout: _SENTINEL | None | float = sentinel,
286 force_close: bool = False,
287 limit: int = 100,
288 limit_per_host: int = 0,
289 enable_cleanup_closed: bool = False,
290 loop: asyncio.AbstractEventLoop | None = None,
291 timeout_ceil_threshold: float = 5,
292 ) -> None:
294 if force_close:
295 if keepalive_timeout is not None and keepalive_timeout is not sentinel:
296 raise ValueError(
297 "keepalive_timeout cannot be set if force_close is True"
298 )
299 keepalive_timeout = None
300 else:
301 if keepalive_timeout is sentinel:
302 keepalive_timeout = 15.0
304 loop = loop or asyncio.get_running_loop()
305 self._timeout_ceil_threshold = timeout_ceil_threshold
307 self._closed = False
308 if loop.get_debug():
309 self._source_traceback = traceback.extract_stack(sys._getframe(1))
311 # Connection pool of reusable connections.
312 # We use a deque to store connections because it has O(1) popleft()
313 # and O(1) append() operations to implement a FIFO queue.
314 self._conns: defaultdict[
315 ConnectionKey, deque[tuple[ResponseHandler, float]]
316 ] = defaultdict(deque)
317 self._limit = limit
318 self._limit_per_host = limit_per_host
319 self._acquired: set[ResponseHandler] = set()
320 self._acquired_per_host: defaultdict[ConnectionKey, set[ResponseHandler]] = (
321 defaultdict(set)
322 )
323 self._keepalive_timeout = keepalive_timeout
324 self._force_close = force_close
326 # {host_key: FIFO list of waiters}
327 # The FIFO is implemented with an OrderedDict with None keys because
328 # python does not have an ordered set.
329 self._waiters: defaultdict[
330 ConnectionKey, OrderedDict[asyncio.Future[None], None]
331 ] = defaultdict(OrderedDict)
333 self._loop = loop
334 self._factory = functools.partial(ResponseHandler, loop=loop)
336 # start keep-alive connection cleanup task
337 self._cleanup_handle: asyncio.TimerHandle | None = None
339 # start cleanup closed transports task
340 self._cleanup_closed_handle: asyncio.TimerHandle | None = None
342 if enable_cleanup_closed and not NEEDS_CLEANUP_CLOSED:
343 warnings.warn(
344 "enable_cleanup_closed ignored because "
345 "https://github.com/python/cpython/pull/118960 is fixed "
346 f"in Python version {sys.version_info}",
347 DeprecationWarning,
348 stacklevel=2,
349 )
350 enable_cleanup_closed = False
352 self._cleanup_closed_disabled = not enable_cleanup_closed
353 self._cleanup_closed_transports: list[asyncio.Transport | None] = []
354 self._placeholder_future: asyncio.Future[Exception | None] = (
355 loop.create_future()
356 )
357 self._placeholder_future.set_result(None)
358 self._cleanup_closed()
360 def __del__(self, _warnings: Any = warnings) -> None:
361 if self._closed:
362 return
363 if not self._conns:
364 return
366 conns = [repr(c) for c in self._conns.values()]
368 self._close()
370 kwargs = {"source": self}
371 _warnings.warn(f"Unclosed connector {self!r}", ResourceWarning, **kwargs)
372 context = {
373 "connector": self,
374 "connections": conns,
375 "message": "Unclosed connector",
376 }
377 if self._source_traceback is not None:
378 context["source_traceback"] = self._source_traceback
379 self._loop.call_exception_handler(context)
381 def __enter__(self) -> "BaseConnector":
382 warnings.warn(
383 '"with Connector():" is deprecated, '
384 'use "async with Connector():" instead',
385 DeprecationWarning,
386 )
387 return self
389 def __exit__(self, *exc: Any) -> None:
390 self._close()
392 async def __aenter__(self) -> "BaseConnector":
393 return self
395 async def __aexit__(
396 self,
397 exc_type: type[BaseException] | None = None,
398 exc_value: BaseException | None = None,
399 exc_traceback: TracebackType | None = None,
400 ) -> None:
401 await self.close()
403 @property
404 def force_close(self) -> bool:
405 """Ultimately close connection on releasing if True."""
406 return self._force_close
408 @property
409 def limit(self) -> int:
410 """The total number for simultaneous connections.
412 If limit is 0 the connector has no limit.
413 The default limit size is 100.
414 """
415 return self._limit
417 @property
418 def limit_per_host(self) -> int:
419 """The limit for simultaneous connections to the same endpoint.
421 Endpoints are the same if they are have equal
422 (host, port, is_ssl) triple.
423 """
424 return self._limit_per_host
426 def _cleanup(self) -> None:
427 """Cleanup unused transports."""
428 if self._cleanup_handle:
429 self._cleanup_handle.cancel()
430 # _cleanup_handle should be unset, otherwise _release() will not
431 # recreate it ever!
432 self._cleanup_handle = None
434 now = monotonic()
435 timeout = self._keepalive_timeout
436 # weakref_handle() never schedules this method for a None timeout
437 assert timeout is not None
439 if self._conns:
440 connections = defaultdict(deque)
441 deadline = now - timeout
442 for key, conns in self._conns.items():
443 alive: deque[tuple[ResponseHandler, float]] = deque()
444 for proto, use_time in conns:
445 if proto.is_connected() and use_time - deadline >= 0:
446 alive.append((proto, use_time))
447 continue
448 transport = proto.transport
449 proto.close()
450 if not self._cleanup_closed_disabled and key.is_ssl:
451 self._cleanup_closed_transports.append(transport)
453 if alive:
454 connections[key] = alive
456 self._conns = connections
458 if self._conns:
459 self._cleanup_handle = helpers.weakref_handle(
460 self,
461 "_cleanup",
462 timeout,
463 self._loop,
464 timeout_ceil_threshold=self._timeout_ceil_threshold,
465 )
467 def _cleanup_closed(self) -> None:
468 """Double confirmation for transport close.
470 Some broken ssl servers may leave socket open without proper close.
471 """
472 if self._cleanup_closed_handle:
473 self._cleanup_closed_handle.cancel()
475 for transport in self._cleanup_closed_transports:
476 if transport is not None:
477 transport.abort()
479 self._cleanup_closed_transports = []
481 if not self._cleanup_closed_disabled:
482 self._cleanup_closed_handle = helpers.weakref_handle(
483 self,
484 "_cleanup_closed",
485 self._cleanup_closed_period,
486 self._loop,
487 timeout_ceil_threshold=self._timeout_ceil_threshold,
488 )
490 def close(self, *, abort_ssl: bool = False) -> Awaitable[None]:
491 """Close all opened transports.
493 :param abort_ssl: If True, SSL connections will be aborted immediately
494 without performing the shutdown handshake. This provides
495 faster cleanup at the cost of less graceful disconnection.
496 """
497 if not (waiters := self._close(abort_ssl=abort_ssl)):
498 # If there are no connections to close, we can return a noop
499 # awaitable to avoid scheduling a task on the event loop.
500 return _DeprecationWaiter(noop())
501 coro = _wait_for_close(waiters)
502 if sys.version_info >= (3, 12):
503 # Optimization for Python 3.12, try to close connections
504 # immediately to avoid having to schedule the task on the event loop.
505 task = asyncio.Task(coro, loop=self._loop, eager_start=True)
506 else:
507 task = self._loop.create_task(coro)
508 return _DeprecationWaiter(task)
510 def _close(self, *, abort_ssl: bool = False) -> list[Awaitable[object]]:
511 waiters: list[Awaitable[object]] = []
513 if self._closed:
514 return waiters
516 self._closed = True
518 try:
519 if self._loop.is_closed():
520 return waiters
522 # cancel cleanup task
523 if self._cleanup_handle:
524 self._cleanup_handle.cancel()
526 # cancel cleanup close task
527 if self._cleanup_closed_handle:
528 self._cleanup_closed_handle.cancel()
530 for data in self._conns.values():
531 for proto, _ in data:
532 if (
533 abort_ssl
534 and proto.transport
535 and proto.transport.get_extra_info("sslcontext") is not None
536 ):
537 proto.abort()
538 else:
539 proto.close()
540 if closed := proto.closed:
541 waiters.append(closed)
543 for proto in self._acquired:
544 if (
545 abort_ssl
546 and proto.transport
547 and proto.transport.get_extra_info("sslcontext") is not None
548 ):
549 proto.abort()
550 else:
551 proto.close()
552 if closed := proto.closed:
553 waiters.append(closed)
555 for transport in self._cleanup_closed_transports:
556 if transport is not None:
557 transport.abort()
559 return waiters
561 finally:
562 self._conns.clear()
563 self._acquired.clear()
564 for keyed_waiters in self._waiters.values():
565 for keyed_waiter in keyed_waiters:
566 keyed_waiter.cancel()
567 self._waiters.clear()
568 self._cleanup_handle = None
569 self._cleanup_closed_transports.clear()
570 self._cleanup_closed_handle = None
572 @property
573 def closed(self) -> bool:
574 """Is connector closed.
576 A readonly property.
577 """
578 return self._closed
580 def _available_connections(self, key: "ConnectionKey") -> int:
581 """
582 Return number of available connections.
584 The limit, limit_per_host and the connection key are taken into account.
586 If it returns less than 1 means that there are no connections
587 available.
588 """
589 # check total available connections
590 # If there are no limits, this will always return 1
591 total_remain = 1
593 if self._limit and (total_remain := self._limit - len(self._acquired)) <= 0:
594 return total_remain
596 # check limit per host
597 if host_remain := self._limit_per_host:
598 if acquired := self._acquired_per_host.get(key):
599 host_remain -= len(acquired)
600 if total_remain > host_remain:
601 return host_remain
603 return total_remain
605 def _update_proxy_auth_header_and_build_proxy_req(
606 self, req: ClientRequest
607 ) -> ClientRequest:
608 """Set Proxy-Authorization header for non-SSL proxy requests and builds the proxy request for SSL proxy requests."""
609 url = req.proxy
610 assert url is not None
611 headers: dict[str, str] = {}
612 if req.proxy_headers is not None:
613 headers = req.proxy_headers # type: ignore[assignment]
614 headers[hdrs.HOST] = req.headers[hdrs.HOST]
615 proxy_req = ClientRequest(
616 hdrs.METH_GET,
617 url,
618 headers=headers,
619 auth=req.proxy_auth,
620 loop=self._loop,
621 ssl=req.ssl,
622 )
623 auth = proxy_req.headers.pop(hdrs.AUTHORIZATION, None)
624 if auth is not None:
625 if not req.is_ssl():
626 req.headers[hdrs.PROXY_AUTHORIZATION] = auth
627 else:
628 proxy_req.headers[hdrs.PROXY_AUTHORIZATION] = auth
629 return proxy_req
631 async def connect(
632 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
633 ) -> Connection:
634 """Get from pool or create new connection."""
635 key = req.connection_key
636 if (conn := await self._get(key, traces)) is not None:
637 # If we do not have to wait and we can get a connection from the pool
638 # we can avoid the timeout ceil logic and directly return the connection
639 if req.proxy:
640 self._update_proxy_auth_header_and_build_proxy_req(req)
641 return conn
643 async with ceil_timeout(timeout.connect, timeout.ceil_threshold):
644 if self._available_connections(key) <= 0:
645 await self._wait_for_available_connection(key, traces)
646 if (conn := await self._get(key, traces)) is not None:
647 if req.proxy:
648 self._update_proxy_auth_header_and_build_proxy_req(req)
649 return conn
651 placeholder = cast(
652 ResponseHandler, _TransportPlaceholder(self._placeholder_future)
653 )
654 self._acquired.add(placeholder)
655 if self._limit_per_host:
656 self._acquired_per_host[key].add(placeholder)
658 try:
659 # Traces are done inside the try block to ensure that the
660 # that the placeholder is still cleaned up if an exception
661 # is raised.
662 if traces:
663 for trace in traces:
664 await trace.send_connection_create_start()
665 proto = await self._create_connection(req, traces, timeout)
666 if traces:
667 for trace in traces:
668 await trace.send_connection_create_end()
669 except BaseException:
670 self._release_acquired(key, placeholder)
671 raise
672 else:
673 if self._closed:
674 proto.close()
675 raise ClientConnectionError("Connector is closed.")
677 # The connection was successfully created, drop the placeholder
678 # and add the real connection to the acquired set. There should
679 # be no awaits after the proto is added to the acquired set
680 # to ensure that the connection is not left in the acquired set
681 # on cancellation.
682 self._acquired.remove(placeholder)
683 self._acquired.add(proto)
684 if self._limit_per_host:
685 acquired_per_host = self._acquired_per_host[key]
686 acquired_per_host.remove(placeholder)
687 acquired_per_host.add(proto)
688 return Connection(self, key, proto, self._loop)
690 async def _wait_for_available_connection(
691 self, key: "ConnectionKey", traces: list["Trace"]
692 ) -> None:
693 """Wait for an available connection slot."""
694 # We loop here because there is a race between
695 # the connection limit check and the connection
696 # being acquired. If the connection is acquired
697 # between the check and the await statement, we
698 # need to loop again to check if the connection
699 # slot is still available.
700 attempts = 0
701 while True:
702 fut: asyncio.Future[None] = self._loop.create_future()
703 keyed_waiters = self._waiters[key]
704 keyed_waiters[fut] = None
705 if attempts:
706 # If we have waited before, we need to move the waiter
707 # to the front of the queue as otherwise we might get
708 # starved and hit the timeout.
709 keyed_waiters.move_to_end(fut, last=False)
711 try:
712 # Traces happen in the try block to ensure that the
713 # the waiter is still cleaned up if an exception is raised.
714 if traces:
715 for trace in traces:
716 await trace.send_connection_queued_start()
717 await fut
718 if traces:
719 for trace in traces:
720 await trace.send_connection_queued_end()
721 finally:
722 # pop the waiter from the queue if its still
723 # there and not already removed by _release_waiter
724 keyed_waiters.pop(fut, None)
725 if not self._waiters.get(key, True):
726 del self._waiters[key]
728 if self._available_connections(key) > 0:
729 break
730 attempts += 1
732 async def _get(
733 self, key: "ConnectionKey", traces: list["Trace"]
734 ) -> Connection | None:
735 """Get next reusable connection for the key or None.
737 The connection will be marked as acquired.
738 """
739 if (conns := self._conns.get(key)) is None:
740 return None
742 t1 = monotonic()
743 while conns:
744 proto, t0 = conns.popleft()
745 # We will we reuse the connection if its connected and
746 # the keepalive timeout has not been exceeded
747 if proto.is_connected() and (
748 self._keepalive_timeout is None or t1 - t0 <= self._keepalive_timeout
749 ):
750 if not conns:
751 # The very last connection was reclaimed: drop the key
752 del self._conns[key]
753 self._acquired.add(proto)
754 if self._limit_per_host:
755 self._acquired_per_host[key].add(proto)
756 if traces:
757 for trace in traces:
758 try:
759 await trace.send_connection_reuseconn()
760 except BaseException:
761 self._release_acquired(key, proto)
762 raise
763 return Connection(self, key, proto, self._loop)
765 # Connection cannot be reused, close it
766 transport = proto.transport
767 proto.close()
768 # only for SSL transports
769 if not self._cleanup_closed_disabled and key.is_ssl:
770 self._cleanup_closed_transports.append(transport)
772 # No more connections: drop the key
773 del self._conns[key]
774 return None
776 def _release_waiter(self) -> None:
777 """
778 Iterates over all waiters until one to be released is found.
780 The one to be released is not finished and
781 belongs to a host that has available connections.
782 """
783 if not self._waiters:
784 return
786 # Having the dict keys ordered this avoids to iterate
787 # at the same order at each call.
788 queues = list(self._waiters)
789 random.shuffle(queues)
791 for key in queues:
792 if self._available_connections(key) < 1:
793 continue
795 waiters = self._waiters[key]
796 while waiters:
797 waiter, _ = waiters.popitem(last=False)
798 if not waiter.done():
799 waiter.set_result(None)
800 return
802 def _release_acquired(self, key: "ConnectionKey", proto: ResponseHandler) -> None:
803 """Release acquired connection."""
804 if self._closed:
805 # acquired connection is already released on connector closing
806 return
808 self._acquired.discard(proto)
809 if self._limit_per_host and (conns := self._acquired_per_host.get(key)):
810 conns.discard(proto)
811 if not conns:
812 del self._acquired_per_host[key]
813 self._release_waiter()
815 def _release(
816 self,
817 key: "ConnectionKey",
818 protocol: ResponseHandler,
819 *,
820 should_close: bool = False,
821 ) -> None:
822 if self._closed:
823 # acquired connection is already released on connector closing
824 return
826 self._release_acquired(key, protocol)
828 if self._force_close or should_close or protocol.should_close:
829 transport = protocol.transport
830 protocol.close()
832 if key.is_ssl and not self._cleanup_closed_disabled:
833 self._cleanup_closed_transports.append(transport)
834 return
836 self._conns[key].append((protocol, monotonic()))
838 if self._cleanup_handle is None:
839 self._cleanup_handle = helpers.weakref_handle(
840 self,
841 "_cleanup",
842 self._keepalive_timeout,
843 self._loop,
844 timeout_ceil_threshold=self._timeout_ceil_threshold,
845 )
847 async def _create_connection(
848 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
849 ) -> ResponseHandler:
850 raise NotImplementedError()
853class _DNSCacheTable:
854 def __init__(self, ttl: float | None = None, max_size: int = 1000) -> None:
855 self._addrs_rr: OrderedDict[
856 tuple[str, int], tuple[Iterator[ResolveResult], int]
857 ] = OrderedDict()
858 self._timestamps: dict[tuple[str, int], float] = {}
859 self._ttl = ttl
860 self._max_size = max_size
862 def __contains__(self, host: object) -> bool:
863 return host in self._addrs_rr
865 def add(self, key: tuple[str, int], addrs: list[ResolveResult]) -> None:
866 if key in self._addrs_rr:
867 self._addrs_rr.move_to_end(key)
869 self._addrs_rr[key] = (cycle(addrs), len(addrs))
871 if self._ttl is not None:
872 self._timestamps[key] = monotonic()
874 if len(self._addrs_rr) > self._max_size:
875 oldest_key, _ = self._addrs_rr.popitem(last=False)
876 self._timestamps.pop(oldest_key, None)
878 def remove(self, key: tuple[str, int]) -> None:
879 self._addrs_rr.pop(key, None)
880 self._timestamps.pop(key, None)
882 def clear(self) -> None:
883 self._addrs_rr.clear()
884 self._timestamps.clear()
886 def next_addrs(self, key: tuple[str, int]) -> list[ResolveResult]:
887 loop, length = self._addrs_rr[key]
888 addrs = list(islice(loop, length))
889 # Consume one more element to shift internal state of `cycle`
890 next(loop)
891 self._addrs_rr.move_to_end(key)
892 return addrs
894 def expired(self, key: tuple[str, int]) -> bool:
895 if self._ttl is None:
896 return False
898 return self._timestamps[key] + self._ttl < monotonic()
901def _make_ssl_context(verified: bool) -> SSLContext:
902 """Create SSL context.
904 This method is not async-friendly and should be called from a thread
905 because it will load certificates from disk and do other blocking I/O.
906 """
907 if ssl is None:
908 # No ssl support
909 return None
910 if verified:
911 sslcontext = ssl.create_default_context()
912 else:
913 sslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
914 sslcontext.options |= ssl.OP_NO_SSLv2
915 sslcontext.options |= ssl.OP_NO_SSLv3
916 sslcontext.check_hostname = False
917 sslcontext.verify_mode = ssl.CERT_NONE
918 sslcontext.options |= ssl.OP_NO_COMPRESSION
919 sslcontext.set_default_verify_paths()
920 sslcontext.set_alpn_protocols(("http/1.1",))
921 return sslcontext
924# The default SSLContext objects are created at import time
925# since they do blocking I/O to load certificates from disk,
926# and imports should always be done before the event loop starts
927# or in a thread.
928_SSL_CONTEXT_VERIFIED = _make_ssl_context(True)
929_SSL_CONTEXT_UNVERIFIED = _make_ssl_context(False)
932class TCPConnector(BaseConnector):
933 """TCP connector.
935 verify_ssl - Set to True to check ssl certifications.
936 fingerprint - Pass the binary sha256
937 digest of the expected certificate in DER format to verify
938 that the certificate the server presents matches. See also
939 https://en.wikipedia.org/wiki/HTTP_Public_Key_Pinning
940 resolver - Enable DNS lookups and use this
941 resolver
942 use_dns_cache - Use memory cache for DNS lookups.
943 ttl_dns_cache - Max seconds having cached a DNS entry, None forever.
944 family - socket address family
945 local_addr - local tuple of (host, port) to bind socket to
947 keepalive_timeout - (optional) Keep-alive timeout. None means a
948 reusable connection never expires from idling alone (a
949 disconnected one is discarded the next time the pool is
950 consulted for that host, not proactively). This is different
951 from force_close=True, which disables reuse entirely.
952 force_close - Set to True to force close and do reconnect
953 after each request (and between redirects).
954 limit - The total number of simultaneous connections.
955 limit_per_host - Number of simultaneous connections to one host.
956 enable_cleanup_closed - Enables clean-up closed ssl transports.
957 Disabled by default.
958 happy_eyeballs_delay - This is the “Connection Attempt Delay”
959 as defined in RFC 8305. To disable
960 the happy eyeballs algorithm, set to None.
961 interleave - “First Address Family Count” as defined in RFC 8305
962 loop - Optional event loop.
963 socket_factory - A SocketFactoryType function that, if supplied,
964 will be used to create sockets given an
965 AddrInfoType.
966 ssl_shutdown_timeout - DEPRECATED. Will be removed in aiohttp 4.0.
967 Grace period for SSL shutdown handshake on TLS
968 connections. Default is 0 seconds (immediate abort).
969 This parameter allowed for a clean SSL shutdown by
970 notifying the remote peer of connection closure,
971 while avoiding excessive delays during connector cleanup.
972 Note: Only takes effect on Python 3.11+.
973 """
975 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"tcp"})
977 def __init__(
978 self,
979 *,
980 verify_ssl: bool = True,
981 fingerprint: bytes | None = None,
982 use_dns_cache: bool = True,
983 ttl_dns_cache: int | None = 10,
984 dns_cache_max_size: int = 1000,
985 family: socket.AddressFamily = socket.AddressFamily.AF_UNSPEC,
986 ssl_context: SSLContext | None = None,
987 ssl: bool | Fingerprint | SSLContext = True,
988 local_addr: tuple[str, int] | None = None,
989 resolver: AbstractResolver | None = None,
990 keepalive_timeout: None | float | _SENTINEL = sentinel,
991 force_close: bool = False,
992 limit: int = 100,
993 limit_per_host: int = 0,
994 enable_cleanup_closed: bool = False,
995 loop: asyncio.AbstractEventLoop | None = None,
996 timeout_ceil_threshold: float = 5,
997 happy_eyeballs_delay: float | None = 0.25,
998 interleave: int | None = None,
999 socket_factory: SocketFactoryType | None = None,
1000 ssl_shutdown_timeout: _SENTINEL | None | float = sentinel,
1001 ):
1002 super().__init__(
1003 keepalive_timeout=keepalive_timeout,
1004 force_close=force_close,
1005 limit=limit,
1006 limit_per_host=limit_per_host,
1007 enable_cleanup_closed=enable_cleanup_closed,
1008 loop=loop,
1009 timeout_ceil_threshold=timeout_ceil_threshold,
1010 )
1012 self._ssl = _merge_ssl_params(ssl, verify_ssl, ssl_context, fingerprint)
1014 self._resolver: AbstractResolver
1015 if resolver is None:
1016 self._resolver = DefaultResolver(loop=self._loop)
1017 self._resolver_owner = True
1018 else:
1019 self._resolver = resolver
1020 self._resolver_owner = False
1022 self._use_dns_cache = use_dns_cache
1023 self._cached_hosts = _DNSCacheTable(
1024 ttl=ttl_dns_cache, max_size=dns_cache_max_size
1025 )
1026 self._throttle_dns_futures: dict[tuple[str, int], set[asyncio.Future[None]]] = (
1027 {}
1028 )
1029 self._family = family
1030 self._local_addr_infos = aiohappyeyeballs.addr_to_addr_infos(local_addr)
1031 self._happy_eyeballs_delay = happy_eyeballs_delay
1032 self._interleave = interleave
1033 self._resolve_host_tasks: set[asyncio.Task[list[ResolveResult]]] = set()
1034 self._socket_factory = socket_factory
1035 self._ssl_shutdown_timeout: float | None
1036 # Handle ssl_shutdown_timeout with warning for Python < 3.11
1037 if ssl_shutdown_timeout is sentinel:
1038 self._ssl_shutdown_timeout = 0
1039 else:
1040 # Deprecation warning for ssl_shutdown_timeout parameter
1041 warnings.warn(
1042 "The ssl_shutdown_timeout parameter is deprecated and will be removed in aiohttp 4.0",
1043 DeprecationWarning,
1044 stacklevel=2,
1045 )
1046 if (
1047 sys.version_info < (3, 11)
1048 and ssl_shutdown_timeout is not None
1049 and ssl_shutdown_timeout != 0
1050 ):
1051 warnings.warn(
1052 f"ssl_shutdown_timeout={ssl_shutdown_timeout} is ignored on Python < 3.11; "
1053 "only ssl_shutdown_timeout=0 is supported. The timeout will be ignored.",
1054 RuntimeWarning,
1055 stacklevel=2,
1056 )
1057 self._ssl_shutdown_timeout = ssl_shutdown_timeout
1059 def _close(self, *, abort_ssl: bool = False) -> list[Awaitable[object]]:
1060 """Close all ongoing DNS calls."""
1061 for fut in chain.from_iterable(self._throttle_dns_futures.values()):
1062 fut.cancel()
1064 waiters = super()._close(abort_ssl=abort_ssl)
1066 for t in self._resolve_host_tasks:
1067 t.cancel()
1068 waiters.append(t)
1070 return waiters
1072 async def close(self, *, abort_ssl: bool = False) -> None:
1073 """
1074 Close all opened transports.
1076 :param abort_ssl: If True, SSL connections will be aborted immediately
1077 without performing the shutdown handshake. If False (default),
1078 the behavior is determined by ssl_shutdown_timeout:
1079 - If ssl_shutdown_timeout=0: connections are aborted
1080 - If ssl_shutdown_timeout>0: graceful shutdown is performed
1081 """
1082 # Use abort_ssl param if explicitly set, otherwise use ssl_shutdown_timeout default
1083 await super().close(abort_ssl=abort_ssl or self._ssl_shutdown_timeout == 0)
1084 if self._resolver_owner:
1085 await self._resolver.close()
1087 @property
1088 def family(self) -> int:
1089 """Socket family like AF_INET."""
1090 return self._family
1092 @property
1093 def use_dns_cache(self) -> bool:
1094 """True if local DNS caching is enabled."""
1095 return self._use_dns_cache
1097 def clear_dns_cache(self, host: str | None = None, port: int | None = None) -> None:
1098 """Remove specified host/port or clear all dns local cache."""
1099 if host is not None and port is not None:
1100 self._cached_hosts.remove((host, port))
1101 elif host is not None or port is not None:
1102 raise ValueError("either both host and port or none of them are allowed")
1103 else:
1104 self._cached_hosts.clear()
1106 async def _resolve_host(
1107 self, host: str, port: int, traces: Sequence["Trace"] | None = None
1108 ) -> list[ResolveResult]:
1109 """Resolve host and return list of addresses."""
1110 if is_ip_address(host):
1111 # Reject legacy numeric IPv4 forms (e.g. 2130706433, 127.1) that
1112 # socket would map onto an address, slipping past a connector-level
1113 # policy that only sees the raw host.
1114 if ":" not in host and not is_canonical_ipv4_address(host):
1115 raise InvalidUrlClientError(host, "is not a canonical IPv4 address")
1116 return [
1117 {
1118 "hostname": host,
1119 "host": host,
1120 "port": port,
1121 "family": self._family,
1122 "proto": 0,
1123 "flags": 0,
1124 }
1125 ]
1127 if not self._use_dns_cache:
1129 if traces:
1130 for trace in traces:
1131 await trace.send_dns_resolvehost_start(host)
1133 if self._closed:
1134 raise ClientConnectionError("Connector is closed")
1136 res = await self._resolver.resolve(host, port, family=self._family)
1138 if traces:
1139 for trace in traces:
1140 await trace.send_dns_resolvehost_end(host)
1142 return res
1144 key = (host, port)
1145 if key in self._cached_hosts and not self._cached_hosts.expired(key):
1146 # get result early, before any await (#4014)
1147 result = self._cached_hosts.next_addrs(key)
1149 if traces:
1150 for trace in traces:
1151 await trace.send_dns_cache_hit(host)
1152 return result
1154 futures: set[asyncio.Future[None]]
1155 #
1156 # If multiple connectors are resolving the same host, we wait
1157 # for the first one to resolve and then use the result for all of them.
1158 # We use a throttle to ensure that we only resolve the host once
1159 # and then use the result for all the waiters.
1160 #
1161 if key in self._throttle_dns_futures:
1162 # get futures early, before any await (#4014)
1163 futures = self._throttle_dns_futures[key]
1164 future: asyncio.Future[None] = self._loop.create_future()
1165 futures.add(future)
1166 if traces:
1167 for trace in traces:
1168 await trace.send_dns_cache_hit(host)
1169 try:
1170 await future
1171 finally:
1172 futures.discard(future)
1173 return self._cached_hosts.next_addrs(key)
1175 # update dict early, before any await (#4014)
1176 self._throttle_dns_futures[key] = futures = set()
1177 # In this case we need to create a task to ensure that we can shield
1178 # the task from cancellation as cancelling this lookup should not cancel
1179 # the underlying lookup or else the cancel event will get broadcast to
1180 # all the waiters across all connections.
1181 #
1182 coro = self._resolve_host_with_throttle(key, host, port, futures, traces)
1183 if sys.version_info >= (3, 14):
1184 # Try to send immediately to avoid having to schedule the task.
1185 loop = asyncio.get_running_loop()
1186 if isinstance(loop, BaseEventLoop):
1187 resolved_host_task = asyncio.create_task(coro, eager_start=True)
1188 else:
1189 resolved_host_task = asyncio.Task(coro, loop=loop, eager_start=True)
1190 elif sys.version_info >= (3, 12):
1191 resolved_host_task = asyncio.Task(
1192 coro, loop=asyncio.get_running_loop(), eager_start=True
1193 )
1194 else:
1195 resolved_host_task = asyncio.create_task(coro)
1197 if not resolved_host_task.done():
1198 self._resolve_host_tasks.add(resolved_host_task)
1199 resolved_host_task.add_done_callback(self._resolve_host_tasks.discard)
1201 try:
1202 return await asyncio.shield(resolved_host_task)
1203 except asyncio.CancelledError:
1205 def drop_exception(fut: "asyncio.Future[list[ResolveResult]]") -> None:
1206 with suppress(Exception, asyncio.CancelledError):
1207 fut.result()
1209 resolved_host_task.add_done_callback(drop_exception)
1210 raise
1212 async def _resolve_host_with_throttle(
1213 self,
1214 key: tuple[str, int],
1215 host: str,
1216 port: int,
1217 futures: set["asyncio.Future[None]"],
1218 traces: Sequence["Trace"] | None,
1219 ) -> list[ResolveResult]:
1220 """Resolve host and set result for all waiters.
1222 This method must be run in a task and shielded from cancellation
1223 to avoid cancelling the underlying lookup.
1224 """
1225 try:
1226 if traces:
1227 for trace in traces:
1228 await trace.send_dns_cache_miss(host)
1230 for trace in traces:
1231 await trace.send_dns_resolvehost_start(host)
1233 addrs = await self._resolver.resolve(host, port, family=self._family)
1234 if traces:
1235 for trace in traces:
1236 await trace.send_dns_resolvehost_end(host)
1238 self._cached_hosts.add(key, addrs)
1239 for fut in futures:
1240 set_result(fut, None)
1241 except BaseException as e:
1242 # any DNS exception is set for the waiters to raise the same exception.
1243 # This coro is always run in task that is shielded from cancellation so
1244 # we should never be propagating cancellation here.
1245 for fut in futures:
1246 set_exception(fut, e)
1247 raise
1248 finally:
1249 self._throttle_dns_futures.pop(key)
1251 return self._cached_hosts.next_addrs(key)
1253 async def _create_connection(
1254 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1255 ) -> ResponseHandler:
1256 """Create connection.
1258 Has same keyword arguments as BaseEventLoop.create_connection.
1259 """
1260 if req.proxy:
1261 _, proto = await self._create_proxy_connection(req, traces, timeout)
1262 else:
1263 _, proto = await self._create_direct_connection(req, traces, timeout)
1265 return proto
1267 def _get_ssl_context(self, req: ClientRequest) -> SSLContext | None:
1268 """Logic to get the correct SSL context
1270 0. if req.ssl is false, return None
1272 1. if ssl_context is specified in req, use it
1273 2. if _ssl_context is specified in self, use it
1274 3. otherwise:
1275 1. if verify_ssl is not specified in req, use self.ssl_context
1276 (will generate a default context according to self.verify_ssl)
1277 2. if verify_ssl is True in req, generate a default SSL context
1278 3. if verify_ssl is False in req, generate a SSL context that
1279 won't verify
1280 """
1281 if not req.is_ssl():
1282 return None
1284 if ssl is None: # pragma: no cover
1285 raise RuntimeError("SSL is not supported.")
1286 sslcontext = req.ssl
1287 if isinstance(sslcontext, ssl.SSLContext):
1288 return sslcontext
1289 if sslcontext is not True:
1290 # not verified or fingerprinted
1291 return _SSL_CONTEXT_UNVERIFIED
1292 sslcontext = self._ssl
1293 if isinstance(sslcontext, ssl.SSLContext):
1294 return sslcontext
1295 if sslcontext is not True:
1296 # not verified or fingerprinted
1297 return _SSL_CONTEXT_UNVERIFIED
1298 return _SSL_CONTEXT_VERIFIED
1300 def _get_fingerprint(self, req: ClientRequest) -> Optional["Fingerprint"]:
1301 ret = req.ssl
1302 if isinstance(ret, Fingerprint):
1303 return ret
1304 ret = self._ssl
1305 if isinstance(ret, Fingerprint):
1306 return ret
1307 return None
1309 async def _wrap_create_connection(
1310 self,
1311 *args: Any,
1312 addr_infos: list[AddrInfoType],
1313 req: ClientRequest,
1314 timeout: "ClientTimeout",
1315 client_error: type[Exception] = ClientConnectorError,
1316 **kwargs: Any,
1317 ) -> tuple[asyncio.Transport, ResponseHandler]:
1318 try:
1319 async with ceil_timeout(
1320 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1321 ):
1322 sock = await aiohappyeyeballs.start_connection(
1323 addr_infos=addr_infos,
1324 local_addr_infos=self._local_addr_infos,
1325 happy_eyeballs_delay=self._happy_eyeballs_delay,
1326 interleave=self._interleave,
1327 loop=self._loop,
1328 socket_factory=self._socket_factory,
1329 )
1330 # Add ssl_shutdown_timeout for Python 3.11+ when SSL is used
1331 if (
1332 kwargs.get("ssl")
1333 and self._ssl_shutdown_timeout
1334 and sys.version_info >= (3, 11)
1335 ):
1336 kwargs["ssl_shutdown_timeout"] = self._ssl_shutdown_timeout
1337 return await self._loop.create_connection(*args, **kwargs, sock=sock)
1338 except cert_errors as exc:
1339 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1340 except ssl_errors as exc:
1341 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1342 except OSError as exc:
1343 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1344 raise
1345 raise client_error(req.connection_key, exc) from exc
1347 async def _wrap_existing_connection(
1348 self,
1349 *args: Any,
1350 req: ClientRequest,
1351 timeout: "ClientTimeout",
1352 client_error: type[Exception] = ClientConnectorError,
1353 **kwargs: Any,
1354 ) -> tuple[asyncio.Transport, ResponseHandler]:
1355 try:
1356 async with ceil_timeout(
1357 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1358 ):
1359 return await self._loop.create_connection(*args, **kwargs)
1360 except cert_errors as exc:
1361 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1362 except ssl_errors as exc:
1363 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1364 except OSError as exc:
1365 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1366 raise
1367 raise client_error(req.connection_key, exc) from exc
1369 def _fail_on_no_start_tls(self, req: "ClientRequest") -> None:
1370 """Raise a :py:exc:`RuntimeError` on missing ``start_tls()``.
1372 It is necessary for TLS-in-TLS so that it is possible to
1373 send HTTPS queries through HTTPS proxies.
1375 This doesn't affect regular HTTP requests, though.
1376 """
1377 if not req.is_ssl():
1378 return
1380 proxy_url = req.proxy
1381 assert proxy_url is not None
1382 if proxy_url.scheme != "https":
1383 return
1385 self._check_loop_for_start_tls()
1387 def _check_loop_for_start_tls(self) -> None:
1388 try:
1389 self._loop.start_tls
1390 except AttributeError as attr_exc:
1391 raise RuntimeError(
1392 "An HTTPS request is being sent through an HTTPS proxy. "
1393 "This needs support for TLS in TLS but it is not implemented "
1394 "in your runtime for the stdlib asyncio.\n\n"
1395 "Please upgrade to Python 3.11 or higher. For more details, "
1396 "please see:\n"
1397 "* https://bugs.python.org/issue37179\n"
1398 "* https://github.com/python/cpython/pull/28073\n"
1399 "* https://docs.aiohttp.org/en/stable/"
1400 "client_advanced.html#proxy-support\n"
1401 "* https://github.com/aio-libs/aiohttp/discussions/6044\n",
1402 ) from attr_exc
1404 def _loop_supports_start_tls(self) -> bool:
1405 try:
1406 self._check_loop_for_start_tls()
1407 except RuntimeError:
1408 return False
1409 else:
1410 return True
1412 def _warn_about_tls_in_tls(
1413 self,
1414 underlying_transport: asyncio.Transport,
1415 req: ClientRequest,
1416 ) -> None:
1417 """Issue a warning if the requested URL has HTTPS scheme."""
1418 if req.request_info.url.scheme != "https":
1419 return
1421 # TLS-in-TLS only applies when the proxy itself is HTTPS.
1422 # When the proxy is HTTP, start_tls upgrades a plain TCP connection,
1423 # which is standard TLS and works on all event loops and Python versions.
1424 if req.proxy is None or req.proxy.scheme != "https":
1425 return
1427 # Check if uvloop is being used, which supports TLS in TLS,
1428 # otherwise assume that asyncio's native transport is being used.
1429 if type(underlying_transport).__module__.startswith("uvloop"):
1430 return
1432 # Support in asyncio was added in Python 3.11 (bpo-44011)
1433 asyncio_supports_tls_in_tls = sys.version_info >= (3, 11) or getattr(
1434 underlying_transport,
1435 "_start_tls_compatible",
1436 False,
1437 )
1439 if asyncio_supports_tls_in_tls:
1440 return
1442 warnings.warn(
1443 "An HTTPS request is being sent through an HTTPS proxy. "
1444 "This support for TLS in TLS is known to be disabled "
1445 "in the stdlib asyncio (Python <3.11). This is why you'll probably see "
1446 "an error in the log below.\n\n"
1447 "It is possible to enable it via monkeypatching. "
1448 "For more details, see:\n"
1449 "* https://bugs.python.org/issue37179\n"
1450 "* https://github.com/python/cpython/pull/28073\n\n"
1451 "You can temporarily patch this as follows:\n"
1452 "* https://docs.aiohttp.org/en/stable/client_advanced.html#proxy-support\n"
1453 "* https://github.com/aio-libs/aiohttp/discussions/6044\n",
1454 RuntimeWarning,
1455 source=self,
1456 # Why `4`? At least 3 of the calls in the stack originate
1457 # from the methods in this class.
1458 stacklevel=3,
1459 )
1461 async def _start_tls_connection(
1462 self,
1463 underlying_transport: asyncio.Transport,
1464 req: ClientRequest,
1465 timeout: "ClientTimeout",
1466 client_error: type[Exception] = ClientConnectorError,
1467 ) -> tuple[asyncio.BaseTransport, ResponseHandler]:
1468 """Wrap the raw TCP transport with TLS."""
1469 tls_proto = self._factory() # Create a brand new proto for TLS
1470 sslcontext = self._get_ssl_context(req)
1471 if TYPE_CHECKING:
1472 # _start_tls_connection is unreachable in the current code path
1473 # if sslcontext is None.
1474 assert sslcontext is not None
1476 try:
1477 async with ceil_timeout(
1478 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1479 ):
1480 try:
1481 # ssl_shutdown_timeout is only available in Python 3.11+
1482 if sys.version_info >= (3, 11) and self._ssl_shutdown_timeout:
1483 tls_transport = await self._loop.start_tls(
1484 underlying_transport,
1485 tls_proto,
1486 sslcontext,
1487 server_hostname=req.server_hostname or req.host,
1488 ssl_handshake_timeout=timeout.total or None,
1489 ssl_shutdown_timeout=self._ssl_shutdown_timeout,
1490 )
1491 else:
1492 tls_transport = await self._loop.start_tls(
1493 underlying_transport,
1494 tls_proto,
1495 sslcontext,
1496 server_hostname=req.server_hostname or req.host,
1497 ssl_handshake_timeout=timeout.total or None,
1498 )
1499 except BaseException:
1500 # We need to close the underlying transport since
1501 # `start_tls()` probably failed before it had a
1502 # chance to do this:
1503 if self._ssl_shutdown_timeout == 0:
1504 underlying_transport.abort()
1505 else:
1506 underlying_transport.close()
1507 raise
1508 if tls_transport is not None:
1509 fingerprint = self._get_fingerprint(req)
1510 if fingerprint:
1511 try:
1512 fingerprint.check(tls_transport)
1513 except ServerFingerprintMismatch:
1514 tls_transport.close()
1515 if not self._cleanup_closed_disabled:
1516 self._cleanup_closed_transports.append(tls_transport)
1517 raise
1518 except cert_errors as exc:
1519 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1520 except ssl_errors as exc:
1521 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1522 except OSError as exc:
1523 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1524 raise
1525 raise client_error(req.connection_key, exc) from exc
1526 except TypeError as type_err:
1527 # Example cause looks like this:
1528 # TypeError: transport <asyncio.sslproto._SSLProtocolTransport
1529 # object at 0x7f760615e460> is not supported by start_tls()
1531 raise ClientConnectionError(
1532 "Cannot initialize a TLS-in-TLS connection to host "
1533 f"{req.host!s}:{req.port:d} through an underlying connection "
1534 f"to an HTTPS proxy {req.proxy!s} ssl:{req.ssl or 'default'} "
1535 f"[{type_err!s}]"
1536 ) from type_err
1537 else:
1538 if tls_transport is None:
1539 msg = "Failed to start TLS (possibly caused by closing transport)"
1540 raise client_error(req.connection_key, OSError(msg))
1541 tls_proto.connection_made(
1542 tls_transport
1543 ) # Kick the state machine of the new TLS protocol
1545 return tls_transport, tls_proto
1547 def _convert_hosts_to_addr_infos(
1548 self, hosts: list[ResolveResult]
1549 ) -> list[AddrInfoType]:
1550 """Converts the list of hosts to a list of addr_infos.
1552 The list of hosts is the result of a DNS lookup. The list of
1553 addr_infos is the result of a call to `socket.getaddrinfo()`.
1554 """
1555 addr_infos: list[AddrInfoType] = []
1556 for hinfo in hosts:
1557 host = hinfo["host"]
1558 is_ipv6 = ":" in host
1559 family = socket.AF_INET6 if is_ipv6 else socket.AF_INET
1560 if self._family and self._family != family:
1561 continue
1562 addr = (host, hinfo["port"], 0, 0) if is_ipv6 else (host, hinfo["port"])
1563 addr_infos.append(
1564 (family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", addr)
1565 )
1566 return addr_infos
1568 async def _create_direct_connection(
1569 self,
1570 req: ClientRequest,
1571 traces: list["Trace"],
1572 timeout: "ClientTimeout",
1573 *,
1574 client_error: type[Exception] = ClientConnectorError,
1575 ) -> tuple[asyncio.Transport, ResponseHandler]:
1576 sslcontext = self._get_ssl_context(req)
1577 fingerprint = self._get_fingerprint(req)
1579 host = req.url.raw_host
1580 assert host is not None
1581 # Replace multiple trailing dots with a single one.
1582 # A trailing dot is only present for fully-qualified domain names.
1583 # See https://github.com/aio-libs/aiohttp/pull/7364.
1584 if host.endswith(".."):
1585 host = host.rstrip(".") + "."
1586 port = req.port
1587 assert port is not None
1588 try:
1589 # Cancelling this lookup should not cancel the underlying lookup
1590 # or else the cancel event will get broadcast to all the waiters
1591 # across all connections.
1592 hosts = await self._resolve_host(host, port, traces=traces)
1593 except OSError as exc:
1594 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1595 raise
1596 # in case of proxy it is not ClientProxyConnectionError
1597 # it is problem of resolving proxy ip itself
1598 raise ClientConnectorDNSError(req.connection_key, exc) from exc
1600 last_exc: Exception | None = None
1601 addr_infos = self._convert_hosts_to_addr_infos(hosts)
1602 while addr_infos:
1603 # Strip trailing dots, certificates contain FQDN without dots.
1604 # See https://github.com/aio-libs/aiohttp/issues/3636
1605 server_hostname = (
1606 (req.server_hostname or host).rstrip(".") if sslcontext else None
1607 )
1609 try:
1610 transp, proto = await self._wrap_create_connection(
1611 self._factory,
1612 timeout=timeout,
1613 ssl=sslcontext,
1614 addr_infos=addr_infos,
1615 server_hostname=server_hostname,
1616 req=req,
1617 client_error=client_error,
1618 )
1619 except (ClientConnectorError, asyncio.TimeoutError) as exc:
1620 last_exc = exc
1621 aiohappyeyeballs.pop_addr_infos_interleave(addr_infos, self._interleave)
1622 continue
1624 if req.is_ssl() and fingerprint:
1625 try:
1626 fingerprint.check(transp)
1627 except ServerFingerprintMismatch as exc:
1628 transp.close()
1629 if not self._cleanup_closed_disabled:
1630 self._cleanup_closed_transports.append(transp)
1631 last_exc = exc
1632 # Remove the bad peer from the list of addr_infos
1633 sock: socket.socket = transp.get_extra_info("socket")
1634 bad_peer = sock.getpeername()
1635 aiohappyeyeballs.remove_addr_infos(addr_infos, bad_peer)
1636 continue
1638 return transp, proto
1639 else:
1640 assert last_exc is not None
1641 raise last_exc
1643 async def _create_proxy_connection(
1644 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1645 ) -> tuple[asyncio.BaseTransport, ResponseHandler]:
1646 self._fail_on_no_start_tls(req)
1647 runtime_has_start_tls = self._loop_supports_start_tls()
1648 proxy_req = self._update_proxy_auth_header_and_build_proxy_req(req)
1650 # create connection to proxy server
1651 transport, proto = await self._create_direct_connection(
1652 proxy_req, [], timeout, client_error=ClientProxyConnectionError
1653 )
1655 if req.is_ssl():
1656 if runtime_has_start_tls:
1657 self._warn_about_tls_in_tls(transport, req)
1659 # For HTTPS requests over HTTP proxy
1660 # we must notify proxy to tunnel connection
1661 # so we send CONNECT command:
1662 # CONNECT www.python.org:443 HTTP/1.1
1663 # Host: www.python.org
1664 #
1665 # next we must do TLS handshake and so on
1666 # to do this we must wrap raw socket into secure one
1667 # asyncio handles this perfectly
1668 proxy_req.method = hdrs.METH_CONNECT
1669 proxy_req.url = req.url
1670 key = req.connection_key._replace(
1671 proxy=None, proxy_auth=None, proxy_headers_hash=None
1672 )
1673 conn = _ConnectTunnelConnection(self, key, proto, self._loop)
1674 try:
1675 proxy_resp = await proxy_req.send(conn)
1676 except BaseException:
1677 conn.close()
1678 raise
1679 try:
1680 protocol = conn._protocol
1681 assert protocol is not None
1683 # read_until_eof=True will ensure the connection isn't closed
1684 # once the response is received and processed allowing
1685 # START_TLS to work on the connection below.
1686 protocol.set_response_params(
1687 read_until_eof=runtime_has_start_tls,
1688 timeout_ceil_threshold=self._timeout_ceil_threshold,
1689 )
1690 resp = await proxy_resp.start(conn)
1691 except BaseException:
1692 proxy_resp.close()
1693 conn.close()
1694 raise
1695 else:
1696 conn._protocol = None
1697 try:
1698 if resp.status != 200:
1699 message = resp.reason
1700 if message is None:
1701 message = HTTPStatus(resp.status).phrase
1702 raise ClientHttpProxyError(
1703 proxy_resp.request_info,
1704 resp.history,
1705 status=resp.status,
1706 message=message,
1707 headers=resp.headers,
1708 )
1709 if not runtime_has_start_tls:
1710 rawsock = transport.get_extra_info("socket", default=None)
1711 if rawsock is None:
1712 raise RuntimeError(
1713 "Transport does not expose socket instance"
1714 )
1715 # Duplicate the socket, so now we can close proxy transport
1716 rawsock = rawsock.dup()
1717 except BaseException:
1718 # It shouldn't be closed in `finally` because it's fed to
1719 # `loop.start_tls()` and the docs say not to touch it after
1720 # passing there.
1721 transport.close()
1722 raise
1723 finally:
1724 if not runtime_has_start_tls:
1725 transport.close()
1727 if not runtime_has_start_tls:
1728 # HTTP proxy with support for upgrade to HTTPS
1729 sslcontext = self._get_ssl_context(req)
1730 return await self._wrap_existing_connection(
1731 self._factory,
1732 timeout=timeout,
1733 ssl=sslcontext,
1734 sock=rawsock,
1735 server_hostname=req.host,
1736 req=req,
1737 )
1739 return await self._start_tls_connection(
1740 # Access the old transport for the last time before it's
1741 # closed and forgotten forever:
1742 transport,
1743 req=req,
1744 timeout=timeout,
1745 )
1746 finally:
1747 proxy_resp.close()
1749 return transport, proto
1752class UnixConnector(BaseConnector):
1753 """Unix socket connector.
1755 path - Unix socket path.
1756 keepalive_timeout - (optional) Keep-alive timeout. None means a
1757 reusable connection never expires from idling alone (a
1758 disconnected one is discarded the next time the pool is
1759 consulted for that host, not proactively). This is different
1760 from force_close=True, which disables reuse entirely.
1761 force_close - Set to True to force close and do reconnect
1762 after each request (and between redirects).
1763 limit - The total number of simultaneous connections.
1764 limit_per_host - Number of simultaneous connections to one host.
1765 loop - Optional event loop.
1766 """
1768 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"unix"})
1770 def __init__(
1771 self,
1772 path: str,
1773 force_close: bool = False,
1774 keepalive_timeout: _SENTINEL | float | None = sentinel,
1775 limit: int = 100,
1776 limit_per_host: int = 0,
1777 loop: asyncio.AbstractEventLoop | None = None,
1778 ) -> None:
1779 super().__init__(
1780 force_close=force_close,
1781 keepalive_timeout=keepalive_timeout,
1782 limit=limit,
1783 limit_per_host=limit_per_host,
1784 loop=loop,
1785 )
1786 self._path = path
1788 @property
1789 def path(self) -> str:
1790 """Path to unix socket."""
1791 return self._path
1793 async def _create_connection(
1794 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1795 ) -> ResponseHandler:
1796 try:
1797 async with ceil_timeout(
1798 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1799 ):
1800 _, proto = await self._loop.create_unix_connection(
1801 self._factory, self._path
1802 )
1803 except OSError as exc:
1804 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1805 raise
1806 raise UnixClientConnectorError(self.path, req.connection_key, exc) from exc
1808 return proto
1811class NamedPipeConnector(BaseConnector):
1812 """Named pipe connector.
1814 Only supported by the proactor event loop.
1815 See also: https://docs.python.org/3/library/asyncio-eventloop.html
1817 path - Windows named pipe path.
1818 keepalive_timeout - (optional) Keep-alive timeout. None means a
1819 reusable connection never expires from idling alone (a
1820 disconnected one is discarded the next time the pool is
1821 consulted for that host, not proactively). This is different
1822 from force_close=True, which disables reuse entirely.
1823 force_close - Set to True to force close and do reconnect
1824 after each request (and between redirects).
1825 limit - The total number of simultaneous connections.
1826 limit_per_host - Number of simultaneous connections to one host.
1827 loop - Optional event loop.
1828 """
1830 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"npipe"})
1832 def __init__(
1833 self,
1834 path: str,
1835 force_close: bool = False,
1836 keepalive_timeout: _SENTINEL | float | None = sentinel,
1837 limit: int = 100,
1838 limit_per_host: int = 0,
1839 loop: asyncio.AbstractEventLoop | None = None,
1840 ) -> None:
1841 super().__init__(
1842 force_close=force_close,
1843 keepalive_timeout=keepalive_timeout,
1844 limit=limit,
1845 limit_per_host=limit_per_host,
1846 loop=loop,
1847 )
1848 if not isinstance(
1849 self._loop,
1850 asyncio.ProactorEventLoop, # type: ignore[attr-defined]
1851 ):
1852 raise RuntimeError(
1853 "Named Pipes only available in proactor loop under windows"
1854 )
1855 self._path = path
1857 @property
1858 def path(self) -> str:
1859 """Path to the named pipe."""
1860 return self._path
1862 async def _create_connection(
1863 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1864 ) -> ResponseHandler:
1865 try:
1866 async with ceil_timeout(
1867 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1868 ):
1869 _, proto = await self._loop.create_pipe_connection( # type: ignore[attr-defined]
1870 self._factory, self._path
1871 )
1872 # the drain is required so that the connection_made is called
1873 # and transport is set otherwise it is not set before the
1874 # `assert conn.transport is not None`
1875 # in client.py's _request method
1876 await asyncio.sleep(0)
1877 # other option is to manually set transport like
1878 # `proto.transport = trans`
1879 except OSError as exc:
1880 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1881 raise
1882 raise ClientConnectorError(req.connection_key, exc) from exc
1884 return cast(ResponseHandler, proto)