1 from __future__ import annotations 2 3 import itertools 4 import logging 5 import ssl 6 import types 7 import typing 8 9 from .._backends.auto import AutoBackend 10 from .._backends.base import SOCKET_OPTION, AsyncNetworkBackend, AsyncNetworkStream 11 from .._exceptions import ConnectError, ConnectTimeout 12 from .._models import Origin, Request, Response 13 from .._ssl import default_ssl_context 14 from .._synchronization import AsyncLock 15 from .._trace import Trace 16 from .http11 import AsyncHTTP11Connection 17 from .interfaces import AsyncConnectionInterface 18 19 RETRIES_BACKOFF_FACTOR = 0.5 # 0s, 0.5s, 1s, 2s, 4s, etc. 20 21 22 logger = logging.getLogger("httpcore.connection") 23 24 25 def exponential_backoff(factor: float) -> typing.Iterator[float]: 26 """ 27 Generate a geometric sequence that has a ratio of 2 and starts with 0. 28 29 For example: 30 - `factor = 2`: `0, 2, 4, 8, 16, 32, 64, ...` 31 - `factor = 3`: `0, 3, 6, 12, 24, 48, 96, ...` 32 """ 33 yield 0 34 for n in itertools.count(): 35 yield factor * 2**n 36 37 38 class AsyncHTTPConnection(AsyncConnectionInterface): 39 def __init__( 40 self, 41 origin: Origin, 42 ssl_context: ssl.SSLContext | None = None, 43 keepalive_expiry: float | None = None, 44 http1: bool = True, 45 http2: bool = False, 46 retries: int = 0, 47 local_address: str | None = None, 48 uds: str | None = None, 49 network_backend: AsyncNetworkBackend | None = None, 50 socket_options: typing.Iterable[SOCKET_OPTION] | None = None, 51 ) -> None: 52 self._origin = origin 53 self._ssl_context = ssl_context 54 self._keepalive_expiry = keepalive_expiry 55 self._http1 = http1 56 self._http2 = http2 57 self._retries = retries 58 self._local_address = local_address 59 self._uds = uds 60 61 self._network_backend: AsyncNetworkBackend = ( 62 AutoBackend() if network_backend is None else network_backend 63 ) 64 self._connection: AsyncConnectionInterface | None = None 65 self._connect_failed: bool = False 66 self._request_lock = AsyncLock() 67 self._socket_options = socket_options 68 69 async def handle_async_request(self, request: Request) -> Response: 70 if not self.can_handle_request(request.url.origin): 71 raise RuntimeError( 72 f"Attempted to send request to {request.url.origin} on connection to {self._origin}" 73 ) 74 75 try: 76 async with self._request_lock: 77 if self._connection is None: 78 stream = await self._connect(request) 79 80 ssl_object = stream.get_extra_info("ssl_object") 81 http2_negotiated = ( 82 ssl_object is not None 83 and ssl_object.selected_alpn_protocol() == "h2" 84 ) 85 if http2_negotiated or (self._http2 and not self._http1): 86 from .http2 import AsyncHTTP2Connection 87 88 self._connection = AsyncHTTP2Connection( 89 origin=self._origin, 90 stream=stream, 91 keepalive_expiry=self._keepalive_expiry, 92 ) 93 else: 94 self._connection = AsyncHTTP11Connection( 95 origin=self._origin, 96 stream=stream, 97 keepalive_expiry=self._keepalive_expiry, 98 ) 99 except BaseException as exc: 100 self._connect_failed = True 101 raise exc 102 103 return await self._connection.handle_async_request(request) 104 105 async def _connect(self, request: Request) -> AsyncNetworkStream: 106 timeouts = request.extensions.get("timeout", {}) 107 sni_hostname = request.extensions.get("sni_hostname", None) 108 timeout = timeouts.get("connect", None) 109 110 retries_left = self._retries 111 delays = exponential_backoff(factor=RETRIES_BACKOFF_FACTOR) 112 113 while True: 114 try: 115 if self._uds is None: 116 kwargs = { 117 "host": self._origin.host.decode("ascii"), 118 "port": self._origin.port, 119 "local_address": self._local_address, 120 "timeout": timeout, 121 "socket_options": self._socket_options, 122 } 123 async with Trace("connect_tcp", logger, request, kwargs) as trace: 124 stream = await self._network_backend.connect_tcp(**kwargs) 125 trace.return_value = stream 126 else: 127 kwargs = { 128 "path": self._uds, 129 "timeout": timeout, 130 "socket_options": self._socket_options, 131 } 132 async with Trace( 133 "connect_unix_socket", logger, request, kwargs 134 ) as trace: 135 stream = await self._network_backend.connect_unix_socket( 136 **kwargs 137 ) 138 trace.return_value = stream 139 140 if self._origin.scheme in (b"https", b"wss"): 141 ssl_context = ( 142 default_ssl_context() 143 if self._ssl_context is None 144 else self._ssl_context 145 ) 146 alpn_protocols = ["http/1.1", "h2"] if self._http2 else ["http/1.1"] 147 ssl_context.set_alpn_protocols(alpn_protocols) 148 149 kwargs = { 150 "ssl_context": ssl_context, 151 "server_hostname": sni_hostname 152 or self._origin.host.decode("ascii"), 153 "timeout": timeout, 154 } 155 async with Trace("start_tls", logger, request, kwargs) as trace: 156 stream = await stream.start_tls(**kwargs) 157 trace.return_value = stream 158 return stream 159 except (ConnectError, ConnectTimeout): 160 if retries_left <= 0: 161 raise 162 retries_left -= 1 163 delay = next(delays) 164 async with Trace("retry", logger, request, kwargs) as trace: 165 await self._network_backend.sleep(delay) 166 167 def can_handle_request(self, origin: Origin) -> bool: 168 return origin == self._origin 169 170 async def aclose(self) -> None: 171 if self._connection is not None: 172 async with Trace("close", logger, None, {}): 173 await self._connection.aclose() 174 175 def is_available(self) -> bool: 176 if self._connection is None: 177 # If HTTP/2 support is enabled, and the resulting connection could 178 # end up as HTTP/2 then we should indicate the connection as being 179 # available to service multiple requests. 180 return ( 181 self._http2 182 and (self._origin.scheme == b"https" or not self._http1) 183 and not self._connect_failed 184 ) 185 return self._connection.is_available() 186 187 def has_expired(self) -> bool: 188 if self._connection is None: 189 return self._connect_failed 190 return self._connection.has_expired() 191 192 def is_idle(self) -> bool: 193 if self._connection is None: 194 return self._connect_failed 195 return self._connection.is_idle() 196 197 def is_closed(self) -> bool: 198 if self._connection is None: 199 return self._connect_failed 200 return self._connection.is_closed() 201 202 def info(self) -> str: 203 if self._connection is None: 204 return "CONNECTION FAILED" if self._connect_failed else "CONNECTING" 205 return self._connection.info() 206 207 def __repr__(self) -> str: 208 return f"<{self.__class__.__name__} [{self.info()}]>" 209 210 # These context managers are not used in the standard flow, but are 211 # useful for testing or working with connection instances directly. 212 213 async def __aenter__(self) -> AsyncHTTPConnection: 214 return self 215 216 async def __aexit__( 217 self, 218 exc_type: type[BaseException] | None = None, 219 exc_value: BaseException | None = None, 220 traceback: types.TracebackType | None = None, 221 ) -> None: 222 await self.aclose() 223