(self, request: Request, timeout: float)
| 408 | self._tls_verbose_logging: bool = tls_verbose_logging |
| 409 | |
| 410 | def _get_agent(self, request: Request, timeout: float) -> Agent: |
| 411 | from twisted.internet import reactor |
| 412 | |
| 413 | bindaddress = request.meta.get("bindaddress") or self._bindAddress |
| 414 | bindaddress = normalize_bind_address(bindaddress) |
| 415 | proxy = request.meta.get("proxy") |
| 416 | if proxy: |
| 417 | proxy = add_http_if_no_scheme(proxy) |
| 418 | proxy_parsed = urlparse(proxy) |
| 419 | proxy_host = proxy_parsed.hostname |
| 420 | proxy_port = proxy_parsed.port |
| 421 | if not proxy_port: |
| 422 | proxy_port = 443 if proxy_parsed.scheme == "https" else 80 |
| 423 | if urlparse_cached(request).scheme == "https": |
| 424 | if proxy_parsed.scheme == "https": # pragma: no cover |
| 425 | raise NotImplementedError( |
| 426 | "HTTPS proxies for HTTPS destinations are not supported" |
| 427 | ) |
| 428 | assert proxy_host is not None |
| 429 | proxyAuth = request.headers.get(b"Proxy-Authorization", None) |
| 430 | proxyConf = (proxy_host, proxy_port, proxyAuth) |
| 431 | return _TunnelingAgent( |
| 432 | reactor=reactor, |
| 433 | proxyConf=proxyConf, |
| 434 | contextFactory=self._contextFactory, |
| 435 | connectTimeout=timeout, |
| 436 | bindAddress=bindaddress, |
| 437 | pool=self._pool, |
| 438 | ) |
| 439 | return _ScrapyProxyAgent( |
| 440 | reactor=reactor, |
| 441 | proxyURI=to_bytes(proxy, encoding="ascii"), |
| 442 | contextFactory=self._contextFactory, |
| 443 | connectTimeout=timeout, |
| 444 | bindAddress=bindaddress, |
| 445 | pool=self._pool, |
| 446 | ) |
| 447 | |
| 448 | return Agent( |
| 449 | reactor=reactor, |
| 450 | contextFactory=self._contextFactory, |
| 451 | connectTimeout=timeout, |
| 452 | bindAddress=bindaddress, |
| 453 | pool=self._pool, |
| 454 | ) |
| 455 | |
| 456 | def download_request(self, request: Request) -> Deferred[Response]: |
| 457 | from twisted.internet import reactor |
no test coverage detected