Use shared httpx.Client in HTTPXTransport for connection pooling

src/vws/transports.py

@@ -18,6 +18,10 @@ class Transport(Protocol):
     returns a ``Response``.
     """
 
+    def close(self) -> None:
+        """Close the transport and release resources."""
+        ...  # pylint: disable=unnecessary-ellipsis
+
     def __call__(
         self,
         *,
@@ -51,6 +55,13 @@ class RequestsTransport:
     This is the default transport.
     """
 
+    def close(self) -> None:
+        """Close the transport.
+
+        This is a no-op for ``RequestsTransport`` as it does not
+        hold persistent connections.
+        """
+
     def __call__(
         self,
         *,
@@ -97,8 +108,26 @@ class HTTPXTransport:
 
     Use this transport for environments where ``httpx`` is
     preferred over ``requests``.
+    A single ``httpx.Client`` is reused across requests
+    for connection pooling.
     """
 
+    def __init__(self) -> None:
+        """Create an ``HTTPXTransport``."""
+        self._client = httpx.Client()
+
+    def close(self) -> None:
+        """Close the underlying ``httpx.Client``."""
+        self._client.close()
+
+    def __enter__(self) -> Self:
+        """Enter the context manager."""
+        return self
+
+    def __exit__(self, *_args: object) -> None:
+        """Exit the context manager and close the client."""
+        self.close()
+
     def __call__(
         self,
         *,
@@ -136,7 +165,7 @@ def __call__(
                 pool=None,
             )
 
-        httpx_response = httpx.request(
+        httpx_response = self._client.request(
             method=method,
             url=url,
             headers=headers,

tests/test_transports.py

@@ -61,6 +61,28 @@ def test_tuple_timeout() -> None:
         assert isinstance(response, Response)
         assert response.status_code == HTTPStatus.OK
 
+    @staticmethod
+    @respx.mock
+    def test_context_manager() -> None:
+        """``HTTPXTransport`` can be used as a context manager."""
+        route = respx.post(url="https://example.com/test").mock(
+            return_value=httpx.Response(
+                status_code=HTTPStatus.OK,
+                text="OK",
+            ),
+        )
+        with HTTPXTransport() as transport:
+            response = transport(
+                method="POST",
+                url="https://example.com/test",
+                headers={"Content-Type": "text/plain"},
+                data=b"hello",
+                request_timeout=30.0,
+            )
+        assert route.called
+        assert isinstance(response, Response)
+        assert response.status_code == HTTPStatus.OK
+
 
 class TestAsyncHTTPXTransport:
     """Tests for ``AsyncHTTPXTransport``."""