Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/connector.py: 19%
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, cast
18import aiohappyeyeballs
19from aiohappyeyeballs import AddrInfoType, SocketFactoryType
20from multidict import CIMultiDict
22from . import hdrs, helpers
23from .abc import AbstractResolver, ResolveResult
24from .client_exceptions import (
25 ClientConnectionError,
26 ClientConnectorCertificateError,
27 ClientConnectorDNSError,
28 ClientConnectorError,
29 ClientConnectorSSLError,
30 ClientHttpProxyError,
31 ClientProxyConnectionError,
32 InvalidUrlClientError,
33 ServerFingerprintMismatch,
34 UnixClientConnectorError,
35 cert_errors,
36 ssl_errors,
37)
38from .client_proto import ResponseHandler
39from .client_reqrep import (
40 SSL_ALLOWED_TYPES,
41 ClientRequest,
42 ClientRequestBase,
43 Fingerprint,
44)
45from .helpers import (
46 _SENTINEL,
47 HIGH_LEVEL_SCHEMA_SET,
48 ceil_timeout,
49 is_canonical_ipv4_address,
50 is_ip_address,
51 sentinel,
52 set_exception,
53 set_result,
54)
55from .log import client_logger
56from .resolver import DefaultResolver
58try:
59 import aiofastnet
60except ImportError:
61 aiofastnet = None # type: ignore[assignment]
64if sys.version_info >= (3, 12):
65 from collections.abc import Buffer
66else:
67 Buffer = "bytes | bytearray | memoryview[int] | memoryview[bytes]"
69try:
70 import ssl
72 SSLContext = ssl.SSLContext
73except ImportError: # pragma: no cover
74 ssl = None # type: ignore[assignment]
75 SSLContext = object # type: ignore[misc,assignment]
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
102async def create_connection(
103 loop: asyncio.AbstractEventLoop,
104 protocol_factory: Callable[[], ResponseHandler],
105 *,
106 ssl: SSLContext | None,
107 sock: socket.socket,
108 server_hostname: str | None,
109 ssl_shutdown_timeout: float | None = None,
110) -> tuple[asyncio.Transport, ResponseHandler]:
111 if aiofastnet is not None:
112 return await aiofastnet.create_connection(
113 loop,
114 protocol_factory,
115 ssl=ssl,
116 sock=sock,
117 server_hostname=server_hostname,
118 ssl_shutdown_timeout=ssl_shutdown_timeout,
119 )
120 else:
121 return await loop.create_connection( # type: ignore[unreachable]
122 protocol_factory,
123 ssl=ssl,
124 sock=sock,
125 server_hostname=server_hostname,
126 ssl_shutdown_timeout=ssl_shutdown_timeout,
127 )
130async def start_tls(
131 loop: asyncio.AbstractEventLoop,
132 transport: asyncio.Transport,
133 protocol: ResponseHandler,
134 sslcontext: SSLContext,
135 *,
136 server_hostname: str | None,
137 ssl_handshake_timeout: float | None,
138 ssl_shutdown_timeout: float | None = None,
139) -> asyncio.Transport | None:
140 if aiofastnet is not None:
141 return await aiofastnet.start_tls(
142 loop,
143 transport,
144 protocol,
145 sslcontext,
146 server_hostname=server_hostname,
147 ssl_handshake_timeout=ssl_handshake_timeout,
148 ssl_shutdown_timeout=ssl_shutdown_timeout,
149 )
150 else:
151 return await loop.start_tls( # type: ignore[unreachable]
152 transport,
153 protocol,
154 sslcontext,
155 server_hostname=server_hostname,
156 ssl_handshake_timeout=ssl_handshake_timeout,
157 ssl_shutdown_timeout=ssl_shutdown_timeout,
158 )
161class Connection:
162 """Represents a single connection."""
164 __slots__ = (
165 "_key",
166 "_connector",
167 "_loop",
168 "_protocol",
169 "_callbacks",
170 "_source_traceback",
171 )
173 def __init__(
174 self,
175 connector: "BaseConnector",
176 key: "ConnectionKey",
177 protocol: ResponseHandler,
178 loop: asyncio.AbstractEventLoop,
179 ) -> None:
180 self._key = key
181 self._connector = connector
182 self._loop = loop
183 self._protocol: ResponseHandler | None = protocol
184 self._callbacks: list[Callable[[], None]] = []
185 self._source_traceback = (
186 traceback.extract_stack(sys._getframe(1)) if loop.get_debug() else None
187 )
189 def __repr__(self) -> str:
190 return f"Connection<{self._key}>"
192 def __del__(self, _warnings: Any = warnings) -> None:
193 if self._protocol is not None:
194 _warnings.warn(
195 f"Unclosed connection {self!r}", ResourceWarning, source=self
196 )
197 if self._loop.is_closed():
198 return
200 self._connector._release(self._key, self._protocol, should_close=True)
202 context = {"client_connection": self, "message": "Unclosed connection"}
203 if self._source_traceback is not None:
204 context["source_traceback"] = self._source_traceback
205 self._loop.call_exception_handler(context)
207 def __bool__(self) -> Literal[True]:
208 """Force subclasses to not be falsy, to make checks simpler."""
209 return True
211 @property
212 def transport(self) -> asyncio.Transport | None:
213 if self._protocol is None:
214 return None
215 return self._protocol.transport
217 @property
218 def protocol(self) -> ResponseHandler | None:
219 return self._protocol
221 def add_callback(self, callback: Callable[[], None]) -> None:
222 if callback is not None:
223 self._callbacks.append(callback)
225 def _notify_release(self) -> None:
226 callbacks, self._callbacks = self._callbacks[:], []
228 for cb in callbacks:
229 with suppress(Exception):
230 cb()
232 def close(self) -> None:
233 self._notify_release()
235 if self._protocol is not None:
236 self._connector._release(self._key, self._protocol, should_close=True)
237 self._protocol = None
239 def release(self) -> None:
240 self._notify_release()
242 if self._protocol is not None:
243 self._connector._release(self._key, self._protocol)
244 self._protocol = None
246 @property
247 def closed(self) -> bool:
248 return self._protocol is None or not self._protocol.is_connected()
251class _ConnectTunnelConnection(Connection):
252 """Special connection wrapper for CONNECT tunnels that must never be pooled.
254 This connection wraps the proxy connection that will be upgraded with TLS.
255 It must never be released to the pool because:
256 1. Its 'closed' future will never complete, causing session.close() to hang
257 2. It represents an intermediate state, not a reusable connection
258 3. The real connection (with TLS) will be created separately
259 """
261 def release(self) -> None:
262 """Do nothing - don't pool or close the connection.
264 These connections are an intermediate state during the CONNECT tunnel
265 setup and will be cleaned up naturally after the TLS upgrade. If they
266 were to be pooled, they would never be properly closed, causing
267 session.close() to wait forever for their 'closed' future.
268 """
271class _TransportPlaceholder:
272 """placeholder for BaseConnector.connect function"""
274 __slots__ = ("closed", "transport")
276 def __init__(self, closed_future: asyncio.Future[Exception | None]) -> None:
277 """Initialize a placeholder for a transport."""
278 self.closed = closed_future
279 self.transport = None
281 def close(self) -> None:
282 """Close the placeholder."""
284 def abort(self) -> None:
285 """Abort the placeholder (does nothing)."""
288class BaseConnector:
289 """Base connector class.
291 keepalive_timeout - (optional) Keep-alive timeout. None means a
292 reusable connection never expires from idling alone (a
293 disconnected one is discarded the next time the pool is
294 consulted for that host, not proactively). This is different
295 from force_close=True, which disables reuse entirely.
296 force_close - Set to True to force close and do reconnect
297 after each request (and between redirects).
298 limit - The total number of simultaneous connections.
299 limit_per_host - Number of simultaneous connections to one host.
300 enable_cleanup_closed - Enables clean-up closed ssl transports.
301 Disabled by default.
302 timeout_ceil_threshold - Trigger ceiling of timeout values when
303 it's above timeout_ceil_threshold.
304 loop - Optional event loop.
305 """
307 _closed = True # prevent AttributeError in __del__ if ctor was failed
308 _source_traceback = None
310 # abort transport after 2 seconds (cleanup broken connections)
311 _cleanup_closed_period = 2.0
313 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET
315 def __init__(
316 self,
317 *,
318 keepalive_timeout: _SENTINEL | None | float = sentinel,
319 force_close: bool = False,
320 limit: int = 100,
321 limit_per_host: int = 0,
322 enable_cleanup_closed: bool = False,
323 timeout_ceil_threshold: float = 5,
324 ) -> None:
325 if force_close:
326 if keepalive_timeout is not None and keepalive_timeout is not sentinel:
327 raise ValueError(
328 "keepalive_timeout cannot be set if force_close is True"
329 )
330 keepalive_timeout = None
331 else:
332 if keepalive_timeout is sentinel:
333 keepalive_timeout = 15.0
335 self._timeout_ceil_threshold = timeout_ceil_threshold
337 loop = asyncio.get_running_loop()
339 self._closed = False
340 if loop.get_debug():
341 self._source_traceback = traceback.extract_stack(sys._getframe(1))
343 # Connection pool of reusable connections.
344 # We use a deque to store connections because it has O(1) popleft()
345 # and O(1) append() operations to implement a FIFO queue.
346 self._conns: defaultdict[
347 ConnectionKey, deque[tuple[ResponseHandler, float]]
348 ] = defaultdict(deque)
349 self._limit = limit
350 self._limit_per_host = limit_per_host
351 self._acquired: set[ResponseHandler] = set()
352 self._acquired_per_host: defaultdict[ConnectionKey, set[ResponseHandler]] = (
353 defaultdict(set)
354 )
355 self._keepalive_timeout = keepalive_timeout
356 self._force_close = force_close
358 # {host_key: FIFO list of waiters}
359 # The FIFO is implemented with an OrderedDict with None keys because
360 # python does not have an ordered set.
361 self._waiters: defaultdict[
362 ConnectionKey, OrderedDict[asyncio.Future[None], None]
363 ] = defaultdict(OrderedDict)
365 self._loop = loop
366 self._factory = functools.partial(ResponseHandler, loop=loop)
368 # start keep-alive connection cleanup task
369 self._cleanup_handle: asyncio.TimerHandle | None = None
371 # start cleanup closed transports task
372 self._cleanup_closed_handle: asyncio.TimerHandle | None = None
374 if enable_cleanup_closed and not NEEDS_CLEANUP_CLOSED:
375 warnings.warn(
376 "enable_cleanup_closed ignored because "
377 "https://github.com/python/cpython/pull/118960 is fixed "
378 f"in Python version {sys.version_info}",
379 DeprecationWarning,
380 stacklevel=2,
381 )
382 enable_cleanup_closed = False
384 self._cleanup_closed_disabled = not enable_cleanup_closed
385 self._cleanup_closed_transports: list[asyncio.Transport | None] = []
387 self._placeholder_future: asyncio.Future[Exception | None] = (
388 loop.create_future()
389 )
390 self._placeholder_future.set_result(None)
391 self._cleanup_closed()
393 def __del__(self, _warnings: Any = warnings) -> None:
394 if self._closed:
395 return
396 if not self._conns:
397 return
399 conns = [repr(c) for c in self._conns.values()]
401 self._close_immediately()
403 _warnings.warn(f"Unclosed connector {self!r}", ResourceWarning, source=self)
404 context = {
405 "connector": self,
406 "connections": conns,
407 "message": "Unclosed connector",
408 }
409 if self._source_traceback is not None:
410 context["source_traceback"] = self._source_traceback
411 self._loop.call_exception_handler(context)
413 async def __aenter__(self) -> "BaseConnector":
414 return self
416 async def __aexit__(
417 self,
418 exc_type: type[BaseException] | None = None,
419 exc_value: BaseException | None = None,
420 exc_traceback: TracebackType | None = None,
421 ) -> None:
422 await self.close()
424 @property
425 def force_close(self) -> bool:
426 """Ultimately close connection on releasing if True."""
427 return self._force_close
429 @property
430 def limit(self) -> int:
431 """The total number for simultaneous connections.
433 If limit is 0 the connector has no limit.
434 The default limit size is 100.
435 """
436 return self._limit
438 @property
439 def limit_per_host(self) -> int:
440 """The limit for simultaneous connections to the same endpoint.
442 Endpoints are the same if they are have equal
443 (host, port, is_ssl) triple.
444 """
445 return self._limit_per_host
447 def _cleanup(self) -> None:
448 """Cleanup unused transports."""
449 if self._cleanup_handle:
450 self._cleanup_handle.cancel()
451 # _cleanup_handle should be unset, otherwise _release() will not
452 # recreate it ever!
453 self._cleanup_handle = None
455 now = monotonic()
456 timeout = self._keepalive_timeout
457 # weakref_handle() never schedules this method for a None timeout
458 assert timeout is not None
460 if self._conns:
461 connections = defaultdict(deque)
462 deadline = now - timeout
463 for key, conns in self._conns.items():
464 alive: deque[tuple[ResponseHandler, float]] = deque()
465 for proto, use_time in conns:
466 if proto.is_connected() and use_time - deadline >= 0:
467 alive.append((proto, use_time))
468 continue
469 transport = proto.transport
470 proto.close()
471 if not self._cleanup_closed_disabled and key.is_ssl:
472 self._cleanup_closed_transports.append(transport)
474 if alive:
475 connections[key] = alive
477 self._conns = connections
479 if self._conns:
480 self._cleanup_handle = helpers.weakref_handle(
481 self,
482 "_cleanup",
483 timeout,
484 self._loop,
485 timeout_ceil_threshold=self._timeout_ceil_threshold,
486 )
488 def _cleanup_closed(self) -> None:
489 """Double confirmation for transport close.
491 Some broken ssl servers may leave socket open without proper close.
492 """
493 if self._cleanup_closed_handle:
494 self._cleanup_closed_handle.cancel()
496 for transport in self._cleanup_closed_transports:
497 if transport is not None:
498 transport.abort()
500 self._cleanup_closed_transports = []
502 if not self._cleanup_closed_disabled:
503 self._cleanup_closed_handle = helpers.weakref_handle(
504 self,
505 "_cleanup_closed",
506 self._cleanup_closed_period,
507 self._loop,
508 timeout_ceil_threshold=self._timeout_ceil_threshold,
509 )
511 async def close(self, *, abort_ssl: bool = False) -> None:
512 """Close all opened transports.
514 :param abort_ssl: If True, SSL connections will be aborted immediately
515 without performing the shutdown handshake. This provides
516 faster cleanup at the cost of less graceful disconnection.
517 """
518 waiters = self._close_immediately(abort_ssl=abort_ssl)
519 if waiters:
520 results = await asyncio.gather(*waiters, return_exceptions=True)
521 for res in results:
522 if isinstance(res, Exception):
523 err_msg = "Error while closing connector: " + repr(res)
524 client_logger.debug(err_msg)
526 def _close_immediately(self, *, abort_ssl: bool = False) -> list[Awaitable[object]]:
527 waiters: list[Awaitable[object]] = []
529 if self._closed:
530 return waiters
532 self._closed = True
534 try:
535 if self._loop.is_closed():
536 return waiters
538 # cancel cleanup task
539 if self._cleanup_handle:
540 self._cleanup_handle.cancel()
542 # cancel cleanup close task
543 if self._cleanup_closed_handle:
544 self._cleanup_closed_handle.cancel()
546 for data in self._conns.values():
547 for proto, _ in data:
548 if (
549 abort_ssl
550 and proto.transport
551 and proto.transport.get_extra_info("sslcontext") is not None
552 ):
553 proto.abort()
554 else:
555 proto.close()
556 if closed := proto.closed:
557 waiters.append(closed)
559 for proto in self._acquired:
560 if (
561 abort_ssl
562 and proto.transport
563 and proto.transport.get_extra_info("sslcontext") is not None
564 ):
565 proto.abort()
566 else:
567 proto.close()
568 if closed := proto.closed:
569 waiters.append(closed)
571 # TODO (A.Yushovskiy, 24-May-2019) collect transp. closing futures
572 for transport in self._cleanup_closed_transports:
573 if transport is not None:
574 transport.abort()
576 return waiters
578 finally:
579 self._conns.clear()
580 self._acquired.clear()
581 for keyed_waiters in self._waiters.values():
582 for keyed_waiter in keyed_waiters:
583 keyed_waiter.cancel()
584 self._waiters.clear()
585 self._cleanup_handle = None
586 self._cleanup_closed_transports.clear()
587 self._cleanup_closed_handle = None
589 @property
590 def closed(self) -> bool:
591 """Is connector closed.
593 A readonly property.
594 """
595 return self._closed
597 def _available_connections(self, key: "ConnectionKey") -> int:
598 """
599 Return number of available connections.
601 The limit, limit_per_host and the connection key are taken into account.
603 If it returns less than 1 means that there are no connections
604 available.
605 """
606 # check total available connections
607 # If there are no limits, this will always return 1
608 total_remain = 1
610 if self._limit and (total_remain := self._limit - len(self._acquired)) <= 0:
611 return total_remain
613 # check limit per host
614 if host_remain := self._limit_per_host:
615 if acquired := self._acquired_per_host.get(key):
616 host_remain -= len(acquired)
617 if total_remain > host_remain:
618 return host_remain
620 return total_remain
622 def _update_proxy_auth_header_and_build_proxy_req(
623 self, req: ClientRequest
624 ) -> ClientRequestBase:
625 """Set Proxy-Authorization header for non-SSL proxy requests and builds the proxy request for SSL proxy requests."""
626 url = req.proxy
627 assert url is not None
628 headers = req.proxy_headers or CIMultiDict[str]()
629 headers[hdrs.HOST] = req.headers[hdrs.HOST]
630 proxy_req = ClientRequestBase(
631 hdrs.METH_GET,
632 url,
633 headers=headers,
634 loop=self._loop,
635 ssl=req.ssl,
636 )
637 if not req.is_ssl():
638 # For non-SSL proxies the request goes directly through the proxy,
639 # so any Proxy-Authorization belongs on the request itself, not on
640 # the synthetic proxy request used for SSL CONNECT.
641 proxy_auth = proxy_req.headers.pop(hdrs.PROXY_AUTHORIZATION, None)
642 if proxy_auth is not None:
643 req.headers[hdrs.PROXY_AUTHORIZATION] = proxy_auth
644 return proxy_req
646 async def connect(
647 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
648 ) -> Connection:
649 """Get from pool or create new connection."""
650 key = req.connection_key
651 if (conn := await self._get(key, traces)) is not None:
652 # If we do not have to wait and we can get a connection from the pool
653 # we can avoid the timeout ceil logic and directly return the connection
654 if req.proxy:
655 self._update_proxy_auth_header_and_build_proxy_req(req)
656 return conn
658 async with ceil_timeout(timeout.connect, timeout.ceil_threshold):
659 if self._available_connections(key) <= 0:
660 await self._wait_for_available_connection(key, traces)
661 if (conn := await self._get(key, traces)) is not None:
662 if req.proxy:
663 self._update_proxy_auth_header_and_build_proxy_req(req)
664 return conn
666 placeholder = cast(
667 ResponseHandler, _TransportPlaceholder(self._placeholder_future)
668 )
669 self._acquired.add(placeholder)
670 if self._limit_per_host:
671 self._acquired_per_host[key].add(placeholder)
673 try:
674 # Traces are done inside the try block to ensure that the
675 # that the placeholder is still cleaned up if an exception
676 # is raised.
677 if traces:
678 for trace in traces:
679 await trace.send_connection_create_start()
680 proto = await self._create_connection(req, traces, timeout)
681 if traces:
682 for trace in traces:
683 await trace.send_connection_create_end()
684 except BaseException:
685 self._release_acquired(key, placeholder)
686 raise
687 else:
688 if self._closed:
689 proto.close()
690 raise ClientConnectionError("Connector is closed.")
692 # The connection was successfully created, drop the placeholder
693 # and add the real connection to the acquired set. There should
694 # be no awaits after the proto is added to the acquired set
695 # to ensure that the connection is not left in the acquired set
696 # on cancellation.
697 self._acquired.remove(placeholder)
698 self._acquired.add(proto)
699 if self._limit_per_host:
700 acquired_per_host = self._acquired_per_host[key]
701 acquired_per_host.remove(placeholder)
702 acquired_per_host.add(proto)
703 return Connection(self, key, proto, self._loop)
705 async def _wait_for_available_connection(
706 self, key: "ConnectionKey", traces: list["Trace"]
707 ) -> None:
708 """Wait for an available connection slot."""
709 # We loop here because there is a race between
710 # the connection limit check and the connection
711 # being acquired. If the connection is acquired
712 # between the check and the await statement, we
713 # need to loop again to check if the connection
714 # slot is still available.
715 attempts = 0
716 while True:
717 fut: asyncio.Future[None] = self._loop.create_future()
718 keyed_waiters = self._waiters[key]
719 keyed_waiters[fut] = None
720 if attempts:
721 # If we have waited before, we need to move the waiter
722 # to the front of the queue as otherwise we might get
723 # starved and hit the timeout.
724 keyed_waiters.move_to_end(fut, last=False)
726 try:
727 # Traces happen in the try block to ensure that the
728 # the waiter is still cleaned up if an exception is raised.
729 if traces:
730 for trace in traces:
731 await trace.send_connection_queued_start()
732 await fut
733 if traces:
734 for trace in traces:
735 await trace.send_connection_queued_end()
736 finally:
737 # pop the waiter from the queue if its still
738 # there and not already removed by _release_waiter
739 keyed_waiters.pop(fut, None)
740 if not self._waiters.get(key, True):
741 del self._waiters[key]
743 if self._available_connections(key) > 0:
744 break
745 attempts += 1
747 async def _get(
748 self, key: "ConnectionKey", traces: list["Trace"]
749 ) -> Connection | None:
750 """Get next reusable connection for the key or None.
752 The connection will be marked as acquired.
753 """
754 if (conns := self._conns.get(key)) is None:
755 return None
757 t1 = monotonic()
758 while conns:
759 proto, t0 = conns.popleft()
760 # We will we reuse the connection if its connected and
761 # the keepalive timeout has not been exceeded
762 if proto.is_connected() and (
763 self._keepalive_timeout is None or t1 - t0 <= self._keepalive_timeout
764 ):
765 if not conns:
766 # The very last connection was reclaimed: drop the key
767 del self._conns[key]
768 self._acquired.add(proto)
769 if self._limit_per_host:
770 self._acquired_per_host[key].add(proto)
771 if traces:
772 for trace in traces:
773 try:
774 await trace.send_connection_reuseconn()
775 except BaseException:
776 self._release_acquired(key, proto)
777 raise
778 return Connection(self, key, proto, self._loop)
780 # Connection cannot be reused, close it
781 transport = proto.transport
782 proto.close()
783 # only for SSL transports
784 if not self._cleanup_closed_disabled and key.is_ssl:
785 self._cleanup_closed_transports.append(transport)
787 # No more connections: drop the key
788 del self._conns[key]
789 return None
791 def _release_waiter(self) -> None:
792 """
793 Iterates over all waiters until one to be released is found.
795 The one to be released is not finished and
796 belongs to a host that has available connections.
797 """
798 if not self._waiters:
799 return
801 # Having the dict keys ordered this avoids to iterate
802 # at the same order at each call.
803 queues = list(self._waiters)
804 random.shuffle(queues)
806 for key in queues:
807 if self._available_connections(key) < 1:
808 continue
810 waiters = self._waiters[key]
811 while waiters:
812 waiter, _ = waiters.popitem(last=False)
813 if not waiter.done():
814 waiter.set_result(None)
815 return
817 def _release_acquired(self, key: "ConnectionKey", proto: ResponseHandler) -> None:
818 """Release acquired connection."""
819 if self._closed:
820 # acquired connection is already released on connector closing
821 return
823 self._acquired.discard(proto)
824 if self._limit_per_host and (conns := self._acquired_per_host.get(key)):
825 conns.discard(proto)
826 if not conns:
827 del self._acquired_per_host[key]
828 self._release_waiter()
830 def _release(
831 self,
832 key: "ConnectionKey",
833 protocol: ResponseHandler,
834 *,
835 should_close: bool = False,
836 ) -> None:
837 if self._closed:
838 # acquired connection is already released on connector closing
839 return
841 self._release_acquired(key, protocol)
843 if self._force_close or should_close or protocol.should_close:
844 transport = protocol.transport
845 protocol.close()
846 if key.is_ssl and not self._cleanup_closed_disabled:
847 self._cleanup_closed_transports.append(transport)
848 return
850 self._conns[key].append((protocol, monotonic()))
852 if self._cleanup_handle is None:
853 self._cleanup_handle = helpers.weakref_handle(
854 self,
855 "_cleanup",
856 self._keepalive_timeout,
857 self._loop,
858 timeout_ceil_threshold=self._timeout_ceil_threshold,
859 )
861 async def _create_connection(
862 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
863 ) -> ResponseHandler:
864 raise NotImplementedError()
867class _DNSCacheTable:
868 def __init__(self, ttl: float | None = None, max_size: int = 1000) -> None:
869 self._addrs_rr: OrderedDict[
870 tuple[str, int], tuple[Iterator[ResolveResult], int]
871 ] = OrderedDict()
872 self._timestamps: dict[tuple[str, int], float] = {}
873 self._ttl = ttl
874 self._max_size = max_size
876 def __contains__(self, host: object) -> bool:
877 return host in self._addrs_rr
879 def add(self, key: tuple[str, int], addrs: list[ResolveResult]) -> None:
880 if key in self._addrs_rr:
881 self._addrs_rr.move_to_end(key)
883 self._addrs_rr[key] = (cycle(addrs), len(addrs))
885 if self._ttl is not None:
886 self._timestamps[key] = monotonic()
888 if len(self._addrs_rr) > self._max_size:
889 oldest_key, _ = self._addrs_rr.popitem(last=False)
890 self._timestamps.pop(oldest_key, None)
892 def remove(self, key: tuple[str, int]) -> None:
893 self._addrs_rr.pop(key, None)
894 self._timestamps.pop(key, None)
896 def clear(self) -> None:
897 self._addrs_rr.clear()
898 self._timestamps.clear()
900 def next_addrs(self, key: tuple[str, int]) -> list[ResolveResult]:
901 loop, length = self._addrs_rr[key]
902 addrs = list(islice(loop, length))
903 # Consume one more element to shift internal state of `cycle`
904 next(loop)
905 self._addrs_rr.move_to_end(key)
906 return addrs
908 def expired(self, key: tuple[str, int]) -> bool:
909 if self._ttl is None:
910 return False
912 return self._timestamps[key] + self._ttl < monotonic()
915def _make_ssl_context(verified: bool) -> SSLContext:
916 """Create SSL context.
918 This method is not async-friendly and should be called from a thread
919 because it will load certificates from disk and do other blocking I/O.
920 """
921 if ssl is None:
922 # No ssl support
923 return None # type: ignore[unreachable]
924 if verified:
925 sslcontext = ssl.create_default_context()
926 else:
927 sslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
928 sslcontext.options |= ssl.OP_NO_SSLv2
929 sslcontext.options |= ssl.OP_NO_SSLv3
930 sslcontext.check_hostname = False
931 sslcontext.verify_mode = ssl.CERT_NONE
932 sslcontext.options |= ssl.OP_NO_COMPRESSION
933 sslcontext.set_default_verify_paths()
934 sslcontext.set_alpn_protocols(("http/1.1",))
935 return sslcontext
938# The default SSLContext objects are created at import time
939# since they do blocking I/O to load certificates from disk,
940# and imports should always be done before the event loop starts
941# or in a thread.
942_SSL_CONTEXT_VERIFIED = _make_ssl_context(True)
943_SSL_CONTEXT_UNVERIFIED = _make_ssl_context(False)
946class TCPConnector(BaseConnector):
947 """TCP connector.
949 ssl - DEPRECATED. Will be removed in aiohttp 5.0.
950 SSL validation mode: ``True`` for the default checks, ``False`` to
951 skip certificate validation, a Fingerprint for certificate pinning
952 or an ssl.SSLContext for custom validation. Pass ``ssl`` to
953 ClientSession or to the individual request instead.
954 resolver - Enable DNS lookups and use this
955 resolver
956 use_dns_cache - Use memory cache for DNS lookups.
957 ttl_dns_cache - Max seconds having cached a DNS entry, None forever.
958 family - socket address family
959 local_addr - local tuple of (host, port) to bind socket to
961 keepalive_timeout - (optional) Keep-alive timeout. None means a
962 reusable connection never expires from idling alone (a
963 disconnected one is discarded the next time the pool is
964 consulted for that host, not proactively). This is different
965 from force_close=True, which disables reuse entirely.
966 force_close - Set to True to force close and do reconnect
967 after each request (and between redirects).
968 limit - The total number of simultaneous connections.
969 limit_per_host - Number of simultaneous connections to one host.
970 enable_cleanup_closed - Enables clean-up closed ssl transports.
971 Disabled by default.
972 happy_eyeballs_delay - This is the “Connection Attempt Delay”
973 as defined in RFC 8305. To disable
974 the happy eyeballs algorithm, set to None.
975 interleave - “First Address Family Count” as defined in RFC 8305
976 loop - Optional event loop.
977 socket_factory - A SocketFactoryType function that, if supplied,
978 will be used to create sockets given an
979 AddrInfoType.
980 ssl_shutdown_timeout - DEPRECATED. Will be removed in aiohttp 4.0.
981 Grace period for SSL shutdown handshake on TLS
982 connections. Default is 0 seconds (immediate abort).
983 This parameter allowed for a clean SSL shutdown by
984 notifying the remote peer of connection closure,
985 while avoiding excessive delays during connector cleanup.
986 """
988 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"tcp"})
990 def __init__(
991 self,
992 *,
993 use_dns_cache: bool = True,
994 ttl_dns_cache: int | None = 10,
995 dns_cache_max_size: int = 1000,
996 family: socket.AddressFamily = socket.AddressFamily.AF_UNSPEC,
997 ssl: bool | Fingerprint | SSLContext | _SENTINEL = sentinel,
998 local_addr: tuple[str, int] | None = None,
999 resolver: AbstractResolver | None = None,
1000 keepalive_timeout: None | float | _SENTINEL = sentinel,
1001 force_close: bool = False,
1002 limit: int = 100,
1003 limit_per_host: int = 0,
1004 enable_cleanup_closed: bool = False,
1005 timeout_ceil_threshold: float = 5,
1006 happy_eyeballs_delay: float | None = 0.25,
1007 interleave: int | None = None,
1008 socket_factory: SocketFactoryType | None = None,
1009 ssl_shutdown_timeout: _SENTINEL | None | float = sentinel,
1010 ):
1011 super().__init__(
1012 keepalive_timeout=keepalive_timeout,
1013 force_close=force_close,
1014 limit=limit,
1015 limit_per_host=limit_per_host,
1016 enable_cleanup_closed=enable_cleanup_closed,
1017 timeout_ceil_threshold=timeout_ceil_threshold,
1018 )
1020 self._ssl: bool | Fingerprint | SSLContext
1021 if ssl is sentinel:
1022 self._ssl = True
1023 else:
1024 if not isinstance(ssl, SSL_ALLOWED_TYPES):
1025 raise TypeError(
1026 "ssl should be SSLContext, Fingerprint, or bool, "
1027 f"got {ssl!r} instead."
1028 )
1029 warnings.warn(
1030 "The ssl parameter is deprecated since 4.0 and scheduled for "
1031 "removal in 5.0, pass ssl to ClientSession() or to the "
1032 "individual request instead",
1033 DeprecationWarning,
1034 stacklevel=2,
1035 )
1036 self._ssl = ssl
1038 self._resolver: AbstractResolver
1039 if resolver is None:
1040 self._resolver = DefaultResolver()
1041 self._resolver_owner = True
1042 else:
1043 self._resolver = resolver
1044 self._resolver_owner = False
1046 self._use_dns_cache = use_dns_cache
1047 self._cached_hosts = _DNSCacheTable(
1048 ttl=ttl_dns_cache, max_size=dns_cache_max_size
1049 )
1050 self._throttle_dns_futures: dict[tuple[str, int], set[asyncio.Future[None]]] = (
1051 {}
1052 )
1053 self._family = family
1054 self._local_addr_infos = aiohappyeyeballs.addr_to_addr_infos(local_addr)
1055 self._happy_eyeballs_delay = happy_eyeballs_delay
1056 self._interleave = interleave
1057 self._resolve_host_tasks: set[asyncio.Task[list[ResolveResult]]] = set()
1058 self._socket_factory = socket_factory
1059 self._ssl_shutdown_timeout: float | None
1061 if ssl_shutdown_timeout is sentinel:
1062 self._ssl_shutdown_timeout = 0
1063 else:
1064 # Deprecation warning for ssl_shutdown_timeout parameter
1065 warnings.warn(
1066 "The ssl_shutdown_timeout parameter is deprecated and will be removed in aiohttp 4.0",
1067 DeprecationWarning,
1068 stacklevel=2,
1069 )
1070 self._ssl_shutdown_timeout = ssl_shutdown_timeout
1072 async def close(self, *, abort_ssl: bool = False) -> None:
1073 """Close all opened transports.
1075 :param abort_ssl: If True, SSL connections will be aborted immediately
1076 without performing the shutdown handshake. If False (default),
1077 the behavior is determined by ssl_shutdown_timeout:
1078 - If ssl_shutdown_timeout=0: connections are aborted
1079 - If ssl_shutdown_timeout>0: graceful shutdown is performed
1080 """
1081 # Use abort_ssl param if explicitly set, otherwise use ssl_shutdown_timeout default
1082 await super().close(abort_ssl=abort_ssl or self._ssl_shutdown_timeout == 0)
1083 if self._resolver_owner:
1084 await self._resolver.close()
1086 def _close_immediately(self, *, abort_ssl: bool = False) -> list[Awaitable[object]]:
1087 for fut in chain.from_iterable(self._throttle_dns_futures.values()):
1088 fut.cancel()
1090 waiters = super()._close_immediately(abort_ssl=abort_ssl)
1092 for t in self._resolve_host_tasks:
1093 t.cancel()
1094 waiters.append(t)
1096 return waiters
1098 @property
1099 def family(self) -> int:
1100 """Socket family like AF_INET."""
1101 return self._family
1103 @property
1104 def use_dns_cache(self) -> bool:
1105 """True if local DNS caching is enabled."""
1106 return self._use_dns_cache
1108 def clear_dns_cache(self, host: str | None = None, port: int | None = None) -> None:
1109 """Remove specified host/port or clear all dns local cache."""
1110 if host is not None and port is not None:
1111 self._cached_hosts.remove((host, port))
1112 elif host is not None or port is not None:
1113 raise ValueError("either both host and port or none of them are allowed")
1114 else:
1115 self._cached_hosts.clear()
1117 async def _resolve_host(
1118 self, host: str, port: int, traces: Sequence["Trace"] | None = None
1119 ) -> list[ResolveResult]:
1120 """Resolve host and return list of addresses."""
1121 if is_ip_address(host):
1122 # Reject legacy numeric IPv4 forms (e.g. 2130706433, 127.1) that
1123 # socket would map onto an address, slipping past a connector-level
1124 # policy that only sees the raw host.
1125 if ":" not in host and not is_canonical_ipv4_address(host):
1126 raise InvalidUrlClientError(host, "is not a canonical IPv4 address")
1127 return [
1128 {
1129 "hostname": host,
1130 "host": host,
1131 "port": port,
1132 "family": self._family,
1133 "proto": 0,
1134 "flags": 0,
1135 }
1136 ]
1138 if not self._use_dns_cache:
1139 if traces:
1140 for trace in traces:
1141 await trace.send_dns_resolvehost_start(host)
1143 if self._closed:
1144 raise ClientConnectionError("Connector is closed")
1146 res = await self._resolver.resolve(host, port, family=self._family)
1148 if traces:
1149 for trace in traces:
1150 await trace.send_dns_resolvehost_end(host)
1152 return res
1154 key = (host, port)
1155 if key in self._cached_hosts and not self._cached_hosts.expired(key):
1156 # get result early, before any await (#4014)
1157 result = self._cached_hosts.next_addrs(key)
1159 if traces:
1160 for trace in traces:
1161 await trace.send_dns_cache_hit(host)
1162 return result
1164 futures: set[asyncio.Future[None]]
1165 #
1166 # If multiple connectors are resolving the same host, we wait
1167 # for the first one to resolve and then use the result for all of them.
1168 # We use a throttle to ensure that we only resolve the host once
1169 # and then use the result for all the waiters.
1170 #
1171 if key in self._throttle_dns_futures:
1172 # get futures early, before any await (#4014)
1173 futures = self._throttle_dns_futures[key]
1174 future: asyncio.Future[None] = self._loop.create_future()
1175 futures.add(future)
1176 if traces:
1177 for trace in traces:
1178 await trace.send_dns_cache_hit(host)
1179 try:
1180 await future
1181 finally:
1182 futures.discard(future)
1183 return self._cached_hosts.next_addrs(key)
1185 # update dict early, before any await (#4014)
1186 self._throttle_dns_futures[key] = futures = set()
1187 # In this case we need to create a task to ensure that we can shield
1188 # the task from cancellation as cancelling this lookup should not cancel
1189 # the underlying lookup or else the cancel event will get broadcast to
1190 # all the waiters across all connections.
1191 #
1192 coro = self._resolve_host_with_throttle(key, host, port, futures, traces)
1193 if sys.version_info >= (3, 14):
1194 # Try to send immediately to avoid having to schedule the task.
1195 loop = asyncio.get_running_loop()
1196 if isinstance(loop, BaseEventLoop):
1197 resolved_host_task = asyncio.create_task(coro, eager_start=True)
1198 else:
1199 resolved_host_task = asyncio.Task(coro, loop=loop, eager_start=True)
1200 elif sys.version_info >= (3, 12):
1201 resolved_host_task = asyncio.Task(
1202 coro, loop=asyncio.get_running_loop(), eager_start=True
1203 )
1204 else:
1205 resolved_host_task = asyncio.create_task(coro)
1207 if not resolved_host_task.done():
1208 self._resolve_host_tasks.add(resolved_host_task)
1209 resolved_host_task.add_done_callback(self._resolve_host_tasks.discard)
1211 try:
1212 return await asyncio.shield(resolved_host_task)
1213 except asyncio.CancelledError:
1215 def drop_exception(fut: "asyncio.Future[list[ResolveResult]]") -> None:
1216 with suppress(Exception, asyncio.CancelledError):
1217 fut.result()
1219 resolved_host_task.add_done_callback(drop_exception)
1220 raise
1222 async def _resolve_host_with_throttle(
1223 self,
1224 key: tuple[str, int],
1225 host: str,
1226 port: int,
1227 futures: set[asyncio.Future[None]],
1228 traces: Sequence["Trace"] | None,
1229 ) -> list[ResolveResult]:
1230 """Resolve host and set result for all waiters.
1232 This method must be run in a task and shielded from cancellation
1233 to avoid cancelling the underlying lookup.
1234 """
1235 try:
1236 if traces:
1237 for trace in traces:
1238 await trace.send_dns_cache_miss(host)
1240 for trace in traces:
1241 await trace.send_dns_resolvehost_start(host)
1243 addrs = await self._resolver.resolve(host, port, family=self._family)
1244 if traces:
1245 for trace in traces:
1246 await trace.send_dns_resolvehost_end(host)
1248 self._cached_hosts.add(key, addrs)
1249 for fut in futures:
1250 set_result(fut, None)
1251 except BaseException as e:
1252 # any DNS exception is set for the waiters to raise the same exception.
1253 # This coro is always run in task that is shielded from cancellation so
1254 # we should never be propagating cancellation here.
1255 for fut in futures:
1256 set_exception(fut, e)
1257 raise
1258 finally:
1259 self._throttle_dns_futures.pop(key)
1261 return self._cached_hosts.next_addrs(key)
1263 async def _create_connection(
1264 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1265 ) -> ResponseHandler:
1266 """Create connection.
1268 Has same keyword arguments as BaseEventLoop.create_connection.
1269 """
1270 if req.proxy:
1271 _, proto = await self._create_proxy_connection(req, traces, timeout)
1272 else:
1273 _, proto = await self._create_direct_connection(req, traces, timeout)
1275 return proto
1277 def _get_ssl_context(self, req: ClientRequestBase) -> SSLContext | None:
1278 """Logic to get the correct SSL context
1280 0. if req.ssl is false, return None
1282 1. if ssl_context is specified in req, use it
1283 2. if _ssl_context is specified in self, use it
1284 3. otherwise:
1285 1. if verify_ssl is not specified in req, use self.ssl_context
1286 (will generate a default context according to self.verify_ssl)
1287 2. if verify_ssl is True in req, generate a default SSL context
1288 3. if verify_ssl is False in req, generate a SSL context that
1289 won't verify
1290 """
1291 if not req.is_ssl():
1292 return None
1294 if ssl is None: # pragma: no cover
1295 raise RuntimeError("SSL is not supported.")
1296 sslcontext = req.ssl
1297 if isinstance(sslcontext, ssl.SSLContext):
1298 return sslcontext
1299 if sslcontext is not True:
1300 # not verified or fingerprinted
1301 return _SSL_CONTEXT_UNVERIFIED
1302 sslcontext = self._ssl
1303 if isinstance(sslcontext, ssl.SSLContext):
1304 return sslcontext
1305 if sslcontext is not True:
1306 # not verified or fingerprinted
1307 return _SSL_CONTEXT_UNVERIFIED
1308 return _SSL_CONTEXT_VERIFIED
1310 def _get_fingerprint(self, req: ClientRequestBase) -> "Fingerprint | None":
1311 ret = req.ssl
1312 if isinstance(ret, Fingerprint):
1313 return ret
1314 ret = self._ssl
1315 if isinstance(ret, Fingerprint):
1316 return ret
1317 return None
1319 async def _wrap_create_connection(
1320 self,
1321 *args: Any,
1322 addr_infos: list[AddrInfoType],
1323 req: ClientRequestBase,
1324 timeout: "ClientTimeout",
1325 client_error: type[Exception] = ClientConnectorError,
1326 **kwargs: Any,
1327 ) -> tuple[asyncio.Transport, ResponseHandler]:
1328 try:
1329 async with ceil_timeout(
1330 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1331 ):
1332 sock = await aiohappyeyeballs.start_connection(
1333 addr_infos=addr_infos,
1334 local_addr_infos=self._local_addr_infos,
1335 happy_eyeballs_delay=self._happy_eyeballs_delay,
1336 interleave=self._interleave,
1337 loop=self._loop,
1338 socket_factory=self._socket_factory,
1339 )
1340 if kwargs.get("ssl") and self._ssl_shutdown_timeout:
1341 kwargs["ssl_shutdown_timeout"] = self._ssl_shutdown_timeout
1342 return await create_connection(self._loop, *args, **kwargs, sock=sock)
1343 except cert_errors as exc:
1344 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1345 except ssl_errors as exc:
1346 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1347 except OSError as exc:
1348 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1349 raise
1350 raise client_error(req.connection_key, exc) from exc
1352 async def _start_tls_connection(
1353 self,
1354 underlying_transport: asyncio.Transport,
1355 req: ClientRequest,
1356 timeout: "ClientTimeout",
1357 client_error: type[Exception] = ClientConnectorError,
1358 ) -> tuple[asyncio.BaseTransport, ResponseHandler]:
1359 """Wrap the raw TCP transport with TLS."""
1360 tls_proto = self._factory() # Create a brand new proto for TLS
1361 sslcontext = self._get_ssl_context(req)
1362 if TYPE_CHECKING:
1363 # _start_tls_connection is unreachable in the current code path
1364 # if sslcontext is None.
1365 assert sslcontext is not None
1367 try:
1368 async with ceil_timeout(
1369 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1370 ):
1371 try:
1372 if self._ssl_shutdown_timeout:
1373 tls_transport = await start_tls(
1374 self._loop,
1375 underlying_transport,
1376 tls_proto,
1377 sslcontext,
1378 server_hostname=req.server_hostname or req.url.raw_host,
1379 ssl_handshake_timeout=timeout.total,
1380 ssl_shutdown_timeout=self._ssl_shutdown_timeout,
1381 )
1382 else:
1383 tls_transport = await start_tls(
1384 self._loop,
1385 underlying_transport,
1386 tls_proto,
1387 sslcontext,
1388 server_hostname=req.server_hostname or req.url.raw_host,
1389 ssl_handshake_timeout=timeout.total,
1390 )
1391 except BaseException:
1392 # We need to close the underlying transport since
1393 # `start_tls()` probably failed before it had a
1394 # chance to do this:
1395 if self._ssl_shutdown_timeout == 0:
1396 underlying_transport.abort()
1397 else:
1398 underlying_transport.close()
1399 raise
1400 if tls_transport is not None:
1401 fingerprint = self._get_fingerprint(req)
1402 if fingerprint:
1403 try:
1404 fingerprint.check(tls_transport)
1405 except ServerFingerprintMismatch:
1406 tls_transport.close()
1407 if not self._cleanup_closed_disabled:
1408 self._cleanup_closed_transports.append(tls_transport)
1409 raise
1410 except cert_errors as exc:
1411 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1412 except ssl_errors as exc:
1413 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1414 except OSError as exc:
1415 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1416 raise
1417 raise client_error(req.connection_key, exc) from exc
1418 except TypeError as type_err:
1419 # Example cause looks like this:
1420 # TypeError: transport <asyncio.sslproto._SSLProtocolTransport
1421 # object at 0x7f760615e460> is not supported by start_tls()
1423 raise ClientConnectionError(
1424 "Cannot initialize a TLS-in-TLS connection to host "
1425 f"{req.url.host!s}:{req.url.port:d} through an underlying connection "
1426 f"to an HTTPS proxy {req.proxy!s} ssl:{req.ssl or 'default'} "
1427 f"[{type_err!s}]"
1428 ) from type_err
1429 else:
1430 if tls_transport is None:
1431 msg = "Failed to start TLS (possibly caused by closing transport)"
1432 raise client_error(req.connection_key, OSError(msg))
1433 tls_proto.connection_made(
1434 tls_transport
1435 ) # Kick the state machine of the new TLS protocol
1437 return tls_transport, tls_proto
1439 def _convert_hosts_to_addr_infos(
1440 self, hosts: list[ResolveResult]
1441 ) -> list[AddrInfoType]:
1442 """Converts the list of hosts to a list of addr_infos.
1444 The list of hosts is the result of a DNS lookup. The list of
1445 addr_infos is the result of a call to `socket.getaddrinfo()`.
1446 """
1447 addr_infos: list[AddrInfoType] = []
1448 for hinfo in hosts:
1449 host = hinfo["host"]
1450 is_ipv6 = ":" in host
1451 family = socket.AF_INET6 if is_ipv6 else socket.AF_INET
1452 if self._family and self._family != family:
1453 continue
1454 addr = (host, hinfo["port"], 0, 0) if is_ipv6 else (host, hinfo["port"])
1455 addr_infos.append(
1456 (family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", addr)
1457 )
1458 return addr_infos
1460 async def _create_direct_connection(
1461 self,
1462 req: ClientRequestBase,
1463 traces: list["Trace"],
1464 timeout: "ClientTimeout",
1465 *,
1466 client_error: type[Exception] = ClientConnectorError,
1467 ) -> tuple[asyncio.Transport, ResponseHandler]:
1468 sslcontext = self._get_ssl_context(req)
1469 fingerprint = self._get_fingerprint(req)
1471 host = req.url.raw_host
1472 assert host is not None
1473 # Replace multiple trailing dots with a single one.
1474 # A trailing dot is only present for fully-qualified domain names.
1475 # See https://github.com/aio-libs/aiohttp/pull/7364.
1476 if host.endswith(".."):
1477 host = host.rstrip(".") + "."
1478 port = req.url.port
1479 assert port is not None
1480 try:
1481 # Cancelling this lookup should not cancel the underlying lookup
1482 # or else the cancel event will get broadcast to all the waiters
1483 # across all connections.
1484 hosts = await self._resolve_host(host, port, traces=traces)
1485 except OSError as exc:
1486 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1487 raise
1488 # in case of proxy it is not ClientProxyConnectionError
1489 # it is problem of resolving proxy ip itself
1490 raise ClientConnectorDNSError(req.connection_key, exc) from exc
1492 last_exc: Exception | None = None
1493 addr_infos = self._convert_hosts_to_addr_infos(hosts)
1494 while addr_infos:
1495 # Strip trailing dots, certificates contain FQDN without dots.
1496 # See https://github.com/aio-libs/aiohttp/issues/3636
1497 server_hostname = (
1498 (req.server_hostname or host).rstrip(".") if sslcontext else None
1499 )
1501 try:
1502 transp, proto = await self._wrap_create_connection(
1503 self._factory,
1504 timeout=timeout,
1505 ssl=sslcontext,
1506 addr_infos=addr_infos,
1507 server_hostname=server_hostname,
1508 req=req,
1509 client_error=client_error,
1510 )
1511 except (ClientConnectorError, TimeoutError) as exc:
1512 last_exc = exc
1513 aiohappyeyeballs.pop_addr_infos_interleave(addr_infos, self._interleave)
1514 continue
1516 if req.is_ssl() and fingerprint:
1517 try:
1518 fingerprint.check(transp)
1519 except ServerFingerprintMismatch as exc:
1520 transp.close()
1521 if not self._cleanup_closed_disabled:
1522 self._cleanup_closed_transports.append(transp)
1523 last_exc = exc
1524 # Remove the bad peer from the list of addr_infos
1525 sock: socket.socket = transp.get_extra_info("socket")
1526 bad_peer = sock.getpeername()
1527 aiohappyeyeballs.remove_addr_infos(addr_infos, bad_peer)
1528 continue
1530 return transp, proto
1531 assert last_exc is not None
1532 raise last_exc
1534 async def _create_proxy_connection(
1535 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1536 ) -> tuple[asyncio.BaseTransport, ResponseHandler]:
1537 proxy_req = self._update_proxy_auth_header_and_build_proxy_req(req)
1539 # create connection to proxy server
1540 transport, proto = await self._create_direct_connection(
1541 proxy_req, [], timeout, client_error=ClientProxyConnectionError
1542 )
1544 if req.is_ssl():
1545 # For HTTPS requests over HTTP proxy
1546 # we must notify proxy to tunnel connection
1547 # so we send CONNECT command:
1548 # CONNECT www.python.org:443 HTTP/1.1
1549 # Host: www.python.org
1550 #
1551 # next we must do TLS handshake and so on
1552 # to do this we must wrap raw socket into secure one
1553 # asyncio handles this perfectly
1554 proxy_req.method = hdrs.METH_CONNECT
1555 proxy_req.url = req.url
1556 key = req.connection_key._replace(proxy=None, proxy_headers_hash=None)
1557 conn = _ConnectTunnelConnection(self, key, proto, self._loop)
1558 try:
1559 proxy_resp = await proxy_req._send(conn)
1560 except BaseException:
1561 conn.close()
1562 raise
1563 try:
1564 protocol = conn._protocol
1565 assert protocol is not None
1567 # read_until_eof=True will ensure the connection isn't closed
1568 # once the response is received and processed allowing
1569 # START_TLS to work on the connection below.
1570 protocol.set_response_params(
1571 read_until_eof=True,
1572 timeout_ceil_threshold=self._timeout_ceil_threshold,
1573 )
1574 resp = await proxy_resp.start(conn)
1575 except BaseException:
1576 proxy_resp.close()
1577 conn.close()
1578 raise
1579 else:
1580 conn._protocol = None
1581 try:
1582 if resp.status != 200:
1583 message = resp.reason
1584 if message is None:
1585 message = HTTPStatus(resp.status).phrase
1586 raise ClientHttpProxyError(
1587 proxy_resp.request_info,
1588 resp.history,
1589 status=resp.status,
1590 message=message,
1591 headers=resp.headers,
1592 )
1593 except BaseException:
1594 # It shouldn't be closed in `finally` because it's fed to
1595 # `loop.start_tls()` and the docs say not to touch it after
1596 # passing there.
1597 transport.close()
1598 raise
1600 return await self._start_tls_connection(
1601 # Access the old transport for the last time before it's
1602 # closed and forgotten forever:
1603 transport,
1604 req=req,
1605 timeout=timeout,
1606 )
1607 finally:
1608 proxy_resp.close()
1610 return transport, proto
1613class UnixConnector(BaseConnector):
1614 """Unix socket connector.
1616 path - Unix socket path.
1617 keepalive_timeout - (optional) Keep-alive timeout. None means a
1618 reusable connection never expires from idling alone (a
1619 disconnected one is discarded the next time the pool is
1620 consulted for that host, not proactively). This is different
1621 from force_close=True, which disables reuse entirely.
1622 force_close - Set to True to force close and do reconnect
1623 after each request (and between redirects).
1624 limit - The total number of simultaneous connections.
1625 limit_per_host - Number of simultaneous connections to one host.
1626 loop - Optional event loop.
1627 """
1629 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"unix"})
1631 def __init__(
1632 self,
1633 path: str,
1634 force_close: bool = False,
1635 keepalive_timeout: _SENTINEL | float | None = sentinel,
1636 limit: int = 100,
1637 limit_per_host: int = 0,
1638 ) -> None:
1639 super().__init__(
1640 force_close=force_close,
1641 keepalive_timeout=keepalive_timeout,
1642 limit=limit,
1643 limit_per_host=limit_per_host,
1644 )
1645 self._path = path
1647 @property
1648 def path(self) -> str:
1649 """Path to unix socket."""
1650 return self._path
1652 async def _create_connection(
1653 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1654 ) -> ResponseHandler:
1655 try:
1656 async with ceil_timeout(
1657 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1658 ):
1659 _, proto = await self._loop.create_unix_connection(
1660 self._factory, self._path
1661 )
1662 except OSError as exc:
1663 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1664 raise
1665 raise UnixClientConnectorError(self.path, req.connection_key, exc) from exc
1667 return proto
1670class NamedPipeConnector(BaseConnector):
1671 """Named pipe connector.
1673 Only supported by the proactor event loop.
1674 See also: https://docs.python.org/3/library/asyncio-eventloop.html
1676 path - Windows named pipe path.
1677 keepalive_timeout - (optional) Keep-alive timeout. None means a
1678 reusable connection never expires from idling alone (a
1679 disconnected one is discarded the next time the pool is
1680 consulted for that host, not proactively). This is different
1681 from force_close=True, which disables reuse entirely.
1682 force_close - Set to True to force close and do reconnect
1683 after each request (and between redirects).
1684 limit - The total number of simultaneous connections.
1685 limit_per_host - Number of simultaneous connections to one host.
1686 loop - Optional event loop.
1687 """
1689 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"npipe"})
1691 def __init__(
1692 self,
1693 path: str,
1694 force_close: bool = False,
1695 keepalive_timeout: _SENTINEL | float | None = sentinel,
1696 limit: int = 100,
1697 limit_per_host: int = 0,
1698 ) -> None:
1699 super().__init__(
1700 force_close=force_close,
1701 keepalive_timeout=keepalive_timeout,
1702 limit=limit,
1703 limit_per_host=limit_per_host,
1704 )
1705 if not isinstance(
1706 self._loop,
1707 asyncio.ProactorEventLoop, # type: ignore[attr-defined]
1708 ):
1709 raise RuntimeError(
1710 "Named Pipes only available in proactor loop under windows"
1711 )
1712 self._path = path
1714 @property
1715 def path(self) -> str:
1716 """Path to the named pipe."""
1717 return self._path
1719 async def _create_connection(
1720 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1721 ) -> ResponseHandler:
1722 try:
1723 async with ceil_timeout(
1724 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1725 ):
1726 _, proto = await self._loop.create_pipe_connection( # type: ignore[attr-defined]
1727 self._factory, self._path
1728 )
1729 # the drain is required so that the connection_made is called
1730 # and transport is set otherwise it is not set before the
1731 # `assert conn.transport is not None`
1732 # in client.py's _request method
1733 await asyncio.sleep(0)
1734 # other option is to manually set transport like
1735 # `proto.transport = trans`
1736 except OSError as exc:
1737 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1738 raise
1739 raise ClientConnectorError(req.connection_key, exc) from exc
1741 return cast(ResponseHandler, proto)