Add sync httpx node

Author: sgraaf

docs/sphinx/installation.rst

@@ -11,4 +11,4 @@ Install the ``requests`` package to use :class:`elastic_transport.RequestsHttpNo
 
 Install the ``aiohttp`` package to use :class:`elastic_transport.AiohttpHttpNode`.
 
-Install the ``httpx`` package to use :class:`elastic_transport.HttpxAsyncHttpNode`.
+Install the ``httpx`` package to use :class:`elastic_transport.HttpxHttpNode` or :class:`elastic_transport.HttpxAsyncHttpNode`.

docs/sphinx/nodes.rst

@@ -22,6 +22,9 @@ Node classes
 .. autoclass:: AiohttpHttpNode
    :members:
 
+.. autoclass:: HttpxHttpNode
+   :members:
+
 .. autoclass:: HttpxAsyncHttpNode
    :members:
 

elastic_transport/__init__.py

@@ -37,6 +37,7 @@
     BaseAsyncNode,
     BaseNode,
     HttpxAsyncHttpNode,
+    HttpxHttpNode,
     RequestsHttpNode,
     Urllib3HttpNode,
 )
@@ -74,6 +75,7 @@
     "HeadApiResponse",
     "HttpHeaders",
     "HttpxAsyncHttpNode",
+    "HttpxHttpNode",
     "JsonSerializer",
     "ListApiResponse",
     "NdjsonSerializer",

elastic_transport/_node/__init__.py

@@ -18,7 +18,7 @@
 from ._base import BaseNode, NodeApiResponse
 from ._base_async import BaseAsyncNode
 from ._http_aiohttp import AiohttpHttpNode
-from ._http_httpx import HttpxAsyncHttpNode
+from ._http_httpx import HttpxAsyncHttpNode, HttpxHttpNode
 from ._http_requests import RequestsHttpNode
 from ._http_urllib3 import Urllib3HttpNode
 
@@ -29,5 +29,6 @@
     "NodeApiResponse",
     "RequestsHttpNode",
     "Urllib3HttpNode",
+    "HttpxHttpNode",
     "HttpxAsyncHttpNode",
 ]

elastic_transport/_node/_http_httpx.py

@@ -30,6 +30,7 @@
     BUILTIN_EXCEPTIONS,
     DEFAULT_CA_CERTS,
     RERAISE_EXCEPTIONS,
+    BaseNode,
     NodeApiResponse,
     ssl_context_from_node_config,
 )
@@ -45,6 +46,161 @@
     _HTTPX_META_VERSION = ""
 
 
+class HttpxHttpNode(BaseNode):
+    _CLIENT_META_HTTP_CLIENT = ("hx", _HTTPX_META_VERSION)
+
+    def __init__(self, config: NodeConfig):
+        if not _HTTPX_AVAILABLE:  # pragma: nocover
+            raise ValueError("You must have 'httpx' installed to use HttpxNode")
+        super().__init__(config)
+
+        if config.ssl_assert_fingerprint:
+            raise ValueError(
+                "httpx does not support certificate pinning. https://github.com/encode/httpx/issues/761"
+            )
+
+        ssl_context: Union[ssl.SSLContext, Literal[False]] = False
+        if config.scheme == "https":
+            if config.ssl_context is not None:
+                ssl_context = ssl_context_from_node_config(config)
+            else:
+                ssl_context = ssl_context_from_node_config(config)
+
+                ca_certs = (
+                    DEFAULT_CA_CERTS if config.ca_certs is None else config.ca_certs
+                )
+                if config.verify_certs:
+                    if not ca_certs:
+                        raise ValueError(
+                            "Root certificates are missing for certificate "
+                            "validation. Either pass them in using the ca_certs parameter or "
+                            "install certifi to use it automatically."
+                        )
+                else:
+                    if config.ssl_show_warn:
+                        warnings.warn(
+                            f"Connecting to {self.base_url!r} using TLS with verify_certs=False is insecure",
+                            stacklevel=warn_stacklevel(),
+                            category=SecurityWarning,
+                        )
+
+                if ca_certs is not None:
+                    if os.path.isfile(ca_certs):
+                        ssl_context.load_verify_locations(cafile=ca_certs)
+                    elif os.path.isdir(ca_certs):
+                        ssl_context.load_verify_locations(capath=ca_certs)
+                    else:
+                        raise ValueError("ca_certs parameter is not a path")
+
+                # Use client_cert and client_key variables for SSL certificate configuration.
+                if config.client_cert and not os.path.isfile(config.client_cert):
+                    raise ValueError("client_cert is not a path to a file")
+                if config.client_key and not os.path.isfile(config.client_key):
+                    raise ValueError("client_key is not a path to a file")
+                if config.client_cert and config.client_key:
+                    ssl_context.load_cert_chain(config.client_cert, config.client_key)
+                elif config.client_cert:
+                    ssl_context.load_cert_chain(config.client_cert)
+
+        self.client = httpx.Client(
+            base_url=f"{config.scheme}://{config.host}:{config.port}",
+            limits=httpx.Limits(max_connections=config.connections_per_node),
+            verify=ssl_context or False,
+            timeout=config.request_timeout,
+        )
+
+    def perform_request(
+        self,
+        method: str,
+        target: str,
+        body: Optional[bytes] = None,
+        headers: Optional[HttpHeaders] = None,
+        request_timeout: Union[DefaultType, Optional[float]] = DEFAULT,
+    ) -> NodeApiResponse:
+        resolved_headers = self._headers.copy()
+        if headers:
+            resolved_headers.update(headers)
+
+        if body:
+            if self._http_compress:
+                resolved_body = gzip.compress(body)
+                resolved_headers["content-encoding"] = "gzip"
+            else:
+                resolved_body = body
+        else:
+            resolved_body = None
+
+        try:
+            start = time.perf_counter()
+            if request_timeout is DEFAULT:
+                resp = self.client.request(
+                    method,
+                    target,
+                    content=resolved_body,
+                    headers=dict(resolved_headers),
+                )
+            else:
+                resp = self.client.request(
+                    method,
+                    target,
+                    content=resolved_body,
+                    headers=dict(resolved_headers),
+                    timeout=request_timeout,
+                )
+            response_body = resp.read()
+            duration = time.perf_counter() - start
+        except RERAISE_EXCEPTIONS + BUILTIN_EXCEPTIONS:
+            raise
+        except Exception as e:
+            err: Exception
+            if isinstance(e, (TimeoutError, httpx.TimeoutException)):
+                err = ConnectionTimeout(
+                    "Connection timed out during request", errors=(e,)
+                )
+            elif isinstance(e, ssl.SSLError):
+                err = TlsError(str(e), errors=(e,))
+            # Detect SSL errors for httpx v0.28.0+
+            # Needed until https://github.com/encode/httpx/issues/3350 is fixed
+            elif isinstance(e, httpx.ConnectError) and e.__cause__:
+                context = e.__cause__.__context__
+                if isinstance(context, ssl.SSLError):
+                    err = TlsError(str(context), errors=(e,))
+                else:
+                    err = ConnectionError(str(e), errors=(e,))
+            else:
+                err = ConnectionError(str(e), errors=(e,))
+            self._log_request(
+                method=method,
+                target=target,
+                headers=resolved_headers,
+                body=body,
+                exception=err,
+            )
+            raise err from None
+
+        meta = ApiResponseMeta(
+            resp.status_code,
+            resp.http_version,
+            HttpHeaders(resp.headers),
+            duration,
+            self.config,
+        )
+
+        self._log_request(
+            method=method,
+            target=target,
+            headers=resolved_headers,
+            body=body,
+            meta=meta,
+            response=response_body,
+        )
+
+        return NodeApiResponse(meta, response_body)
+
+    def close(self) -> None:
+        self.client.close()
+
+
 class HttpxAsyncHttpNode(BaseAsyncNode):
     _CLIENT_META_HTTP_CLIENT = ("hx", _HTTPX_META_VERSION)
 

elastic_transport/_transport.py

@@ -56,6 +56,7 @@
     AiohttpHttpNode,
     BaseNode,
     HttpxAsyncHttpNode,
+    HttpxHttpNode,
     RequestsHttpNode,
     Urllib3HttpNode,
 )
@@ -70,6 +71,7 @@
     "urllib3": Urllib3HttpNode,
     "requests": RequestsHttpNode,
     "aiohttp": AiohttpHttpNode,
+    "httpx": HttpxHttpNode,
     "httpxasync": HttpxAsyncHttpNode,
 }
 # These are HTTP status errors that shouldn't be considered

tests/async_/test_async_transport.py

@@ -315,7 +315,7 @@ async def test_node_class_as_string():
         AsyncTransport([NodeConfig("http", "localhost", 80)], node_class="huh?")
     assert str(e.value) == (
         "Unknown option for node_class: 'huh?'. "
-        "Available options are: 'aiohttp', 'httpxasync', 'requests', 'urllib3'"
+        "Available options are: 'aiohttp', 'httpx', 'httpxasync', 'requests', 'urllib3'"
     )
 
 

tests/node/test_base.py

@@ -20,6 +20,7 @@
 from elastic_transport import (
     AiohttpHttpNode,
     HttpxAsyncHttpNode,
+    HttpxHttpNode,
     NodeConfig,
     RequestsHttpNode,
     Urllib3HttpNode,
@@ -28,7 +29,14 @@
 
 
 @pytest.mark.parametrize(
-    "node_cls", [Urllib3HttpNode, RequestsHttpNode, AiohttpHttpNode, HttpxAsyncHttpNode]
+    "node_cls",
+    [
+        Urllib3HttpNode,
+        RequestsHttpNode,
+        AiohttpHttpNode,
+        HttpxHttpNode,
+        HttpxAsyncHttpNode,
+    ],
 )
 def test_unknown_parameter(node_cls):
     with pytest.raises(TypeError):