Implement support for specifying the httpcore Network Backend when creating default transports

Signed-off-by: Patrick J. McNerthney <patrick.mcnerthney@fortra.com>

Author: Patrick J. McNerthney

docs/advanced/transports.md

@@ -35,6 +35,18 @@ connecting via a Unix Domain Socket that is only available via this low-level AP
 {"ID": "...", "Containers": 4, "Images": 74, ...}
 ```
 
+Another advanced configuration is supplying a custom httpcore [Network Backend](https://www.encode.io/httpcore/network-backends/).
+
+```pycon
+>>> import httpx
+>>> import myk8s
+>>> # This custom network backend enables remote access to ports inside a Kubernetes Cluster using pod port forwarding.
+>>> backend = myk8s.NetworkBackend('cluster.local')
+>>> transport = httpx.HTTPTransport(network_backend=backend)
+>>> client = httpx.Client(transport=transport)
+>>> response = client.get("http://argocd-server.argocd.svc.cluster.local")
+```
+
 ## WSGI Transport
 
 You can configure an `httpx` client to call directly into a Python web application using the WSGI protocol.

httpx/_transports/default.py

@@ -6,6 +6,7 @@
 * uds: str
 * local_address: str
 * retries: int
+* network_backend: httpcore.NetworkBackend
 
 Example usages...
 
@@ -22,6 +23,13 @@
 # Using advanced httpcore configuration, with unix domain sockets.
 transport = httpx.HTTPTransport(uds="socket.uds")
 client = httpx.Client(transport=transport)
+
+# Using advanced httpcore configuration, with custom network backend.
+import myk8s
+backend = myk8s.NetworkBackend('cluster.local')
+transport = httpx.HTTPTransport(network_backend=backend)
+client = httpx.Client(transport=transport)
+response = client.get("http://argocd-server.argocd.svc.cluster.local")
 """
 
 from __future__ import annotations
@@ -33,6 +41,8 @@
 if typing.TYPE_CHECKING:
     import ssl  # pragma: no cover
 
+    import httpcore  # pragma: no cover
+
     import httpx  # pragma: no cover
 
 from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
@@ -146,6 +156,7 @@ def __init__(
         local_address: str | None = None,
         retries: int = 0,
         socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
+        network_backend: httpcore.NetworkBackend | None = None,
     ) -> None:
         import httpcore
 
@@ -164,6 +175,7 @@ def __init__(
                 local_address=local_address,
                 retries=retries,
                 socket_options=socket_options,
+                network_backend=network_backend,
             )
         elif proxy.url.scheme in ("http", "https"):
             self._pool = httpcore.HTTPProxy(
@@ -183,6 +195,7 @@ def __init__(
                 http1=http1,
                 http2=http2,
                 socket_options=socket_options,
+                network_backend=network_backend,
             )
         elif proxy.url.scheme in ("socks5", "socks5h"):
             try:
@@ -207,6 +220,7 @@ def __init__(
                 keepalive_expiry=limits.keepalive_expiry,
                 http1=http1,
                 http2=http2,
+                network_backend=network_backend,
             )
         else:  # pragma: no cover
             raise ValueError(
@@ -290,6 +304,7 @@ def __init__(
         local_address: str | None = None,
         retries: int = 0,
         socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
+        network_backend: httpcore.AsyncNetworkBackend | None = None,
     ) -> None:
         import httpcore
 
@@ -308,6 +323,7 @@ def __init__(
                 local_address=local_address,
                 retries=retries,
                 socket_options=socket_options,
+                network_backend=network_backend,
             )
         elif proxy.url.scheme in ("http", "https"):
             self._pool = httpcore.AsyncHTTPProxy(
@@ -327,6 +343,7 @@ def __init__(
                 http1=http1,
                 http2=http2,
                 socket_options=socket_options,
+                network_backend=network_backend,
             )
         elif proxy.url.scheme in ("socks5", "socks5h"):
             try:
@@ -351,6 +368,7 @@ def __init__(
                 keepalive_expiry=limits.keepalive_expiry,
                 http1=http1,
                 http2=http2,
+                network_backend=network_backend,
             )
         else:  # pragma: no cover
             raise ValueError(

tests/client/test_network_backend.py

@@ -0,0 +1,97 @@
+import typing
+
+import httpcore
+import pytest
+
+import httpx
+
+
+def test_network_backend():
+    class Backend(httpcore.NetworkBackend):
+        def connect_tcp(
+            self,
+            host: str,
+            port: int,
+            timeout: typing.Optional[float] = None,
+            local_address: typing.Optional[str] = None,
+            socket_options: typing.Optional[
+                typing.Iterable[httpcore.SOCKET_OPTION]
+            ] = None,
+        ) -> httpcore.NetworkStream:
+            return Stream()
+
+    class Stream(httpcore.NetworkStream):
+        body = b"\r\n".join(
+            [
+                b"HTTP/1.1 200 OK",
+                b"",
+                b"From Backend!",
+            ]
+        )
+
+        def read(self, max_bytes: int, timeout: typing.Optional[float] = None) -> bytes:
+            body = self.body
+            if body:
+                self.body = b""
+            return body
+
+        def write(self, buffer: bytes, timeout: typing.Optional[float] = None) -> None:
+            pass
+
+        def close(self) -> None:
+            pass
+
+    backend = Backend()
+    transport = httpx.HTTPTransport(network_backend=backend)
+    with httpx.Client(transport=transport) as client:
+        response = client.get("http://www.example.org")
+    assert response.status_code == 200
+    assert response.text == "From Backend!"
+
+
+@pytest.mark.anyio
+async def test_async_network_backend():
+    class AsyncBackend(httpcore.AsyncNetworkBackend):
+        async def connect_tcp(
+            self,
+            host: str,
+            port: int,
+            timeout: typing.Optional[float] = None,
+            local_address: typing.Optional[str] = None,
+            socket_options: typing.Optional[
+                typing.Iterable[httpcore.SOCKET_OPTION]
+            ] = None,
+        ) -> httpcore.AsyncNetworkStream:
+            return AsyncStream()
+
+    class AsyncStream(httpcore.AsyncNetworkStream):
+        body = b"\r\n".join(
+            [
+                b"HTTP/1.1 200 OK",
+                b"",
+                b"From Async Backend!",
+            ]
+        )
+
+        async def read(
+            self, max_bytes: int, timeout: typing.Optional[float] = None
+        ) -> bytes:
+            body = self.body
+            if body:
+                self.body = b""
+            return body
+
+        async def write(
+            self, buffer: bytes, timeout: typing.Optional[float] = None
+        ) -> None:
+            pass
+
+        async def aclose(self) -> None:
+            pass
+
+    backend = AsyncBackend()
+    transport = httpx.AsyncHTTPTransport(network_backend=backend)
+    async with httpx.AsyncClient(transport=transport) as client:
+        response = await client.get("http://www.example.org")
+    assert response.status_code == 200
+    assert response.text == "From Async Backend!"