encode/httpx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge eca57b3928 into b5addb64f0

Browse Source
This commit is contained in:
Dan 2026-03-01 03:43:41 -08:00 committed by GitHub
commit 2011b46fea
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 19 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
httpx/_transports/default.py
View File

@ -118,6 +118,15 @@ def map_httpcore_exceptions() -> typing.Iterator[None]:
raise mapped_exc(message) from exc
def create_proxy(proxy: ProxyTypes | None, ssl_context: ssl.SSLContext) -> Proxy | None:
if isinstance(proxy, (str, URL)):
proxy_url = proxy if isinstance(proxy, URL) else URL(proxy)
if proxy_url.scheme == "https":
return Proxy(url=proxy_url, ssl_context=ssl_context)
return Proxy(url=proxy_url)
return proxy
class ResponseStream(SyncByteStream):
def __init__(self, httpcore_stream: typing.Iterable[bytes]) -> None:
self._httpcore_stream = httpcore_stream
@ -149,8 +158,8 @@ class HTTPTransport(BaseTransport):
) -> None:
import httpcore
proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy
ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
proxy = create_proxy(proxy, ssl_context)
if proxy is None:
self._pool = httpcore.ConnectionPool(
@ -293,8 +302,8 @@ class AsyncHTTPTransport(AsyncBaseTransport):
) -> None:
import httpcore
proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy
ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
proxy = create_proxy(proxy, ssl_context)
if proxy is None:
self._pool = httpcore.AsyncConnectionPool(

8
tests/client/test_proxies.py
View File

@ -1,3 +1,5 @@
import ssl
import httpcore
import pytest
@ -263,3 +265,9 @@ def test_proxy_with_mounts():
transport = client._transport_for_url(httpx.URL("http://example.com"))
assert transport == proxy_transport
def test_proxy_with_ssl_context():
ssl_context = ssl.create_default_context()
proxy_transport = httpx.HTTPTransport(proxy="https://127.0.0.1", verify=ssl_context)
assert proxy_transport._pool._proxy_ssl_context == ssl_context # type: ignore[attr-defined]