Retrieve or create an HTTP client for the given proxy URL. If a client for the specified proxy URL does not exist, create and store a new one.
(self, proxy_url: str | None, cookie_jar: CookieJar | None)
| 223 | response.close() |
| 224 | |
| 225 | def _get_client(self, proxy_url: str | None, cookie_jar: CookieJar | None) -> AsyncClient: |
| 226 | """Retrieve or create an HTTP client for the given proxy URL. |
| 227 | |
| 228 | If a client for the specified proxy URL does not exist, create and store a new one. |
| 229 | """ |
| 230 | cached_data = self._client_by_proxy_url.get(proxy_url) |
| 231 | if cached_data: |
| 232 | client = cached_data['client'] |
| 233 | client_cookie_jar = cached_data['cookie_jar'] |
| 234 | if client_cookie_jar is cookie_jar: |
| 235 | # If the cookie jar matches, return the existing client. |
| 236 | return client |
| 237 | |
| 238 | # Prepare a default kwargs for the new client. |
| 239 | kwargs: dict[str, Any] = { |
| 240 | 'proxy': proxy_url, |
| 241 | 'http3': self._http3, |
| 242 | 'verify': self._verify, |
| 243 | 'follow_redirects': True, |
| 244 | 'browser': self._browser, |
| 245 | } |
| 246 | |
| 247 | # Update the default kwargs with any additional user-provided kwargs. |
| 248 | kwargs.update(self._async_client_kwargs) |
| 249 | |
| 250 | client = AsyncClient(**kwargs, cookie_jar=cookie_jar) |
| 251 | |
| 252 | self._client_by_proxy_url[proxy_url] = _ClientCacheEntry(client=client, cookie_jar=cookie_jar) |
| 253 | |
| 254 | return client |
| 255 | |
| 256 | @staticmethod |
| 257 | def _is_proxy_error(error: HTTPError) -> bool: |
no test coverage detected