self.trust_env = trust_env
self.netrc = NetRCInfo()
- def get_proxy_map(
+ def _get_proxy_map(
self, proxies: typing.Optional[ProxiesTypes], trust_env: bool,
) -> typing.Dict[str, Proxy]:
if proxies is None:
trust_env=trust_env,
)
- proxy_map = self.get_proxy_map(proxies, trust_env)
+ proxy_map = self._get_proxy_map(proxies, trust_env)
- self.transport = self.init_transport(
+ self._transport = self._init_transport(
verify=verify,
cert=cert,
http2=http2,
app=app,
trust_env=trust_env,
)
- self.proxies: typing.Dict[str, httpcore.SyncHTTPTransport] = {
- key: self.init_proxy_transport(
+ self._proxies: typing.Dict[str, httpcore.SyncHTTPTransport] = {
+ key: self._init_proxy_transport(
proxy,
verify=verify,
cert=cert,
for key, proxy in proxy_map.items()
}
- def init_transport(
+ def _init_transport(
self,
verify: VerifyTypes = True,
cert: CertTypes = None,
http2=http2,
)
- def init_proxy_transport(
+ def _init_proxy_transport(
self,
proxy: Proxy,
verify: VerifyTypes = True,
http2=http2,
)
- def transport_for_url(self, url: URL) -> httpcore.SyncHTTPTransport:
+ def _transport_for_url(self, url: URL) -> httpcore.SyncHTTPTransport:
"""
Returns the transport instance that should be used for a given URL.
This will either be the standard connection pool, or a proxy.
"""
- if self.proxies and not should_not_be_proxied(url):
+ if self._proxies and not should_not_be_proxied(url):
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 self.proxies:
- transport = self.proxies[proxy_key]
+ if proxy_key and proxy_key in self._proxies:
+ transport = self._proxies[proxy_key]
return transport
- return self.transport
+ return self._transport
def request(
self,
Sends a single request, without handling any redirections.
"""
- transport = self.transport_for_url(request.url)
+ transport = self._transport_for_url(request.url)
try:
with map_exceptions(HTTPCORE_EXC_MAP):
)
def close(self) -> None:
- self.transport.close()
- for proxy in self.proxies.values():
+ self._transport.close()
+ for proxy in self._proxies.values():
proxy.close()
def __enter__(self) -> "Client":
trust_env=trust_env,
)
- proxy_map = self.get_proxy_map(proxies, trust_env)
+ proxy_map = self._get_proxy_map(proxies, trust_env)
- self.transport = self.init_transport(
+ self._transport = self._init_transport(
verify=verify,
cert=cert,
http2=http2,
app=app,
trust_env=trust_env,
)
- self.proxies: typing.Dict[str, httpcore.AsyncHTTPTransport] = {
- key: self.init_proxy_transport(
+ self._proxies: typing.Dict[str, httpcore.AsyncHTTPTransport] = {
+ key: self._init_proxy_transport(
proxy,
verify=verify,
cert=cert,
for key, proxy in proxy_map.items()
}
- def init_transport(
+ def _init_transport(
self,
verify: VerifyTypes = True,
cert: CertTypes = None,
http2=http2,
)
- def init_proxy_transport(
+ def _init_proxy_transport(
self,
proxy: Proxy,
verify: VerifyTypes = True,
http2=http2,
)
- def transport_for_url(self, url: URL) -> httpcore.AsyncHTTPTransport:
+ def _transport_for_url(self, url: URL) -> httpcore.AsyncHTTPTransport:
"""
Returns the transport instance that should be used for a given URL.
This will either be the standard connection pool, or a proxy.
"""
- if self.proxies and not should_not_be_proxied(url):
+ if self._proxies and not should_not_be_proxied(url):
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 self.proxies:
- transport = self.proxies[proxy_key]
+ if proxy_key and proxy_key in self._proxies:
+ transport = self._proxies[proxy_key]
return transport
- return self.transport
+ return self._transport
async def request(
self,
Sends a single request, without handling any redirections.
"""
- transport = self.transport_for_url(request.url)
+ transport = self._transport_for_url(request.url)
try:
with map_exceptions(HTTPCORE_EXC_MAP):
)
async def aclose(self) -> None:
- await self.transport.aclose()
- for proxy in self.proxies.values():
+ await self._transport.aclose()
+ for proxy in self._proxies.values():
await proxy.aclose()
async def __aenter__(self) -> "AsyncClient":
client = httpx.AsyncClient(proxies=proxies)
for proxy_key, url in expected_proxies:
- assert proxy_key in client.proxies
- proxy = client.proxies[proxy_key]
+ assert proxy_key in client._proxies
+ proxy = client._proxies[proxy_key]
assert isinstance(proxy, httpcore.AsyncHTTPProxy)
assert proxy.proxy_origin == httpx.URL(url).raw[:3]
- assert len(expected_proxies) == len(client.proxies)
+ assert len(expected_proxies) == len(client._proxies)
PROXY_URL = "http://[::1]"
)
def test_transport_for_request(url, proxies, expected):
client = httpx.AsyncClient(proxies=proxies)
- transport = client.transport_for_url(httpx.URL(url))
+ transport = client._transport_for_url(httpx.URL(url))
if expected is None:
- assert transport is client.transport
+ assert transport is client._transport
else:
assert isinstance(transport, httpcore.AsyncHTTPProxy)
assert transport.proxy_origin == httpx.URL(expected).raw[:3]
monkeypatch.setenv(name, value)
client = client_class()
- transport = client.transport_for_url(httpx.URL(url))
+ transport = client._transport_for_url(httpx.URL(url))
if expected is None:
- assert transport == client.transport
+ assert transport == client._transport
else:
assert transport.proxy_origin == httpx.URL(expected).raw[:3]