Sends a single request, without handling any redirections.
"""
- dispatcher = self._dispatcher_for_request(request, self.proxies)
+ dispatcher = self.dispatcher_for_url(request.url)
try:
with ElapsedTimer() as timer:
return response
- def _dispatcher_for_request(
- self, request: Request, proxies: typing.Dict[str, Dispatcher]
- ) -> Dispatcher:
- """Gets the Dispatcher instance that should be used for a given Request"""
- if proxies:
- url = request.url
+ def dispatcher_for_url(self, url: URL) -> Dispatcher:
+ """
+ Returns the Dispatcher instance that should be used for a given URL.
+ This will either be the standard connection pool, or a proxy.
+ """
+ if self.proxies:
is_default_port = (url.scheme == "http" and url.port == 80) or (
url.scheme == "https" and url.port == 443
)
"all",
)
for proxy_key in proxy_keys:
- if proxy_key and proxy_key in proxies:
- dispatcher = proxies[proxy_key]
+ if proxy_key and proxy_key in self.proxies:
+ dispatcher = self.proxies[proxy_key]
return dispatcher
return self.dispatch
)
def test_dispatcher_for_request(url, proxies, expected):
client = httpx.Client(proxies=proxies)
- request = httpx.Request("GET", url)
- dispatcher = client._dispatcher_for_request(request, client.proxies)
+ dispatcher = client.dispatcher_for_url(httpx.URL(url))
if expected is None:
assert isinstance(dispatcher, httpx.ConnectionPool)