[HTTPXodus] migrate httpx to httpx2 (hard switch; closes #1856)

langfuse/_client/client.py

@@ -28,7 +28,7 @@
 )
 
 import backoff
-import httpx
+import httpx2
 from opentelemetry import context as otel_context_api
 from opentelemetry import trace as otel_trace_api
 from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
@@ -201,8 +201,8 @@ class Langfuse:
         base_url (Optional[str]): The Langfuse API base URL. Defaults to "https://cloud.langfuse.com". Can also be set via LANGFUSE_BASE_URL environment variable.
         host (Optional[str]): Deprecated. Use base_url instead. The Langfuse API host URL. Defaults to "https://cloud.langfuse.com".
         timeout (Optional[int]): Timeout in seconds for API requests. Defaults to 5 seconds.
-        httpx_client (Optional[httpx.Client]): Custom httpx client for making non-tracing HTTP requests. If not provided, a default client will be created.
-            **Fork safety**: ``httpx.Client`` is thread-safe but not process-safe. When using
+        httpx_client (Optional[httpx2.Client]): Custom httpx2 client for making non-tracing HTTP requests. If not provided, a default client will be created.
+            **Fork safety**: ``httpx2.Client`` is thread-safe but not process-safe. When using
             ``fork()``-based servers (e.g. Gunicorn with ``--preload``), the SDK automatically
             recreates its internally-managed HTTP client in child processes after fork. A custom
             ``httpx_client`` is intentionally left as-is (the fork-inherited copy is reused), so
@@ -318,7 +318,7 @@ def __init__(
         base_url: Optional[str] = None,
         host: Optional[str] = None,
         timeout: Optional[int] = None,
-        httpx_client: Optional[httpx.Client] = None,
+        httpx_client: Optional[httpx2.Client] = None,
         debug: bool = False,
         tracing_enabled: Optional[bool] = True,
         flush_at: Optional[int] = None,
@@ -4219,11 +4219,11 @@ def update_prompt(
         return updated_prompt
 
     def _url_encode(self, url: str, *, is_url_param: Optional[bool] = False) -> str:
-        # httpx ≥ 0.28 does its own WHATWG-compliant quoting (eg. encodes bare
+        # httpx2 ≥ 0.28 does its own WHATWG-compliant quoting (eg. encodes bare
         # “%”, “?”, “#”, “|”, … in query/path parts).  Re-quoting here would
         # double-encode, so we skip when the value is about to be sent straight
-        # to httpx (`is_url_param=True`) and the installed version is ≥ 0.28.
-        if is_url_param and Version(httpx.__version__) >= Version("0.28.0"):
+        # to httpx2 (`is_url_param=True`) and the installed version is ≥ 0.28.
+        if is_url_param and Version(httpx2.__version__) >= Version("0.28.0"):
             return url
 
         # urllib.parse.quote does not escape slashes "/" by default; we need to add safe="" to force escaping

langfuse/_client/resource_manager.py

@@ -23,7 +23,7 @@
 from queue import Full, Queue
 from typing import Any, Callable, Dict, List, Optional, cast
 
-import httpx
+import httpx2
 from opentelemetry import trace as otel_trace_api
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
@@ -88,7 +88,7 @@ class LangfuseResourceManager:
     _ingestion_consumers: List[ScoreIngestionConsumer]
 
     @classmethod
-    def get_singleton_httpx_client(cls) -> Optional[httpx.Client]:
+    def get_singleton_httpx_client(cls) -> Optional[httpx2.Client]:
         with cls._lock:
             instances = list(cls._instances.values())
 
@@ -98,11 +98,11 @@ def get_singleton_httpx_client(cls) -> Optional[httpx.Client]:
             if len(instances) > 1:
                 # Mirror get_client's safety stance: with multiple clients we
                 # cannot tell which one produced a given reference, so fall back
-                # to a default httpx client rather than silently using an
+                # to a default httpx2 client rather than silently using an
                 # arbitrary instance's transport config (proxy / CA / mTLS).
                 langfuse_logger.warning(
                     "Multiple Langfuse clients are instantiated; falling back to a "
-                    "default httpx client for LangfuseMediaReference fetches. Pass an "
+                    "default httpx2 client for LangfuseMediaReference fetches. Pass an "
                     "explicit `client` to fetch_bytes/fetch_base64/fetch_data_uri to "
                     "honor per-client transport settings."
                 )
@@ -121,7 +121,7 @@ def __new__(
         timeout: Optional[int] = None,
         flush_at: Optional[int] = None,
         flush_interval: Optional[float] = None,
-        httpx_client: Optional[httpx.Client] = None,
+        httpx_client: Optional[httpx2.Client] = None,
         media_upload_thread_count: Optional[int] = None,
         sample_rate: Optional[float] = None,
         mask: Optional[MaskFunction] = None,
@@ -189,7 +189,7 @@ def _initialize_instance(
         flush_at: Optional[int] = None,
         flush_interval: Optional[float] = None,
         media_upload_thread_count: Optional[int] = None,
-        httpx_client: Optional[httpx.Client] = None,
+        httpx_client: Optional[httpx2.Client] = None,
         sample_rate: Optional[float] = None,
         mask: Optional[MaskFunction] = None,
         mask_otel_spans: Optional[MaskOtelSpansFunction] = None,
@@ -333,15 +333,15 @@ def _init_media_manager(self) -> None:
     def _init_api_clients(self) -> None:
         """Initialize HTTP-backed API clients.
 
-        Internally-managed httpx clients are recreated when this method is
+        Internally-managed httpx2 clients are recreated when this method is
         called after fork. Caller-provided clients are preserved because their
         lifecycle belongs to the caller.
         """
         if self._custom_httpx_client is not None:
             self.httpx_client = self._custom_httpx_client
         else:
             client_headers = self.additional_headers if self.additional_headers else {}
-            self.httpx_client = httpx.Client(
+            self.httpx_client = httpx2.Client(
                 timeout=self.timeout, headers=client_headers
             )
 
@@ -427,7 +427,7 @@ def _at_fork_reinit(self) -> None:
 
         if sys.platform == "darwin" and not urllib.request.getproxies_environment():
             # urllib proxy discovery falls back to macOS SystemConfiguration APIs that
-            # are not safe to invoke after fork(). Setting no_proxy="*" makes httpx and
+            # are not safe to invoke after fork(). Setting no_proxy="*" makes httpx2 and
             # requests skip that lookup entirely in this child process. Skipped when
             # proxies are configured via environment variables: urllib then never touches
             # SystemConfiguration (no segfault risk), and overriding no_proxy would
@@ -444,13 +444,13 @@ def _at_fork_reinit(self) -> None:
         # belong to the preloaded parent process and must not be processed by every
         # worker — otherwise uploads/scores would be duplicated across workers.
         #
-        # Internally-managed httpx clients must also be recreated: fork() duplicates the
+        # Internally-managed httpx2 clients must also be recreated: fork() duplicates the
         # parent's connection pool (TCP socket file descriptors) into the child. Both
         # processes then share the same underlying sockets, causing data corruption and
         # SSL/TLS state mismatch under concurrent use. Fresh clients start with an empty
         # pool owned solely by this child process.
         #
-        # Custom httpx clients provided by the caller are NOT recreated. The fork-inherited
+        # Custom httpx2 clients provided by the caller are NOT recreated. The fork-inherited
         # copy is reused as-is, giving the caller the opportunity to handle process-safety
         # themselves (e.g. by registering their own os.register_at_fork handler).
         try:

langfuse/_task_manager/media_manager.py

@@ -4,7 +4,7 @@
 from typing import Any, Callable, Optional, TypeVar, cast
 
 import backoff
-import httpx
+import httpx2
 from typing_extensions import ParamSpec
 
 from langfuse._client.environment_variables import LANGFUSE_MEDIA_UPLOAD_ENABLED
@@ -34,7 +34,7 @@ def __init__(
         self,
         *,
         api_client: LangfuseAPI,
-        httpx_client: httpx.Client,
+        httpx_client: httpx2.Client,
         media_upload_queue: Queue,
         max_retries: Optional[int] = 3,
     ):
@@ -50,7 +50,7 @@ def reinitialize(
         self,
         *,
         api_client: LangfuseAPI,
-        httpx_client: httpx.Client,
+        httpx_client: httpx2.Client,
         media_upload_queue: Queue,
     ) -> None:
         self._api_client = api_client
@@ -454,7 +454,7 @@ def _process_upload_media_job(
             headers["x-ms-blob-type"] = "BlockBlob"
             headers["x-amz-checksum-sha256"] = data["content_sha256_hash"]
 
-        def _upload_with_status_check() -> httpx.Response:
+        def _upload_with_status_check() -> httpx2.Response:
             response = self._httpx_client.put(
                 upload_url,
                 headers=headers,
@@ -468,7 +468,7 @@ def _upload_with_status_check() -> httpx.Response:
 
         try:
             upload_response = self._request_with_backoff(_upload_with_status_check)
-        except httpx.HTTPStatusError as e:
+        except httpx2.HTTPStatusError as e:
             upload_time_ms = int((time.time() - upload_start_time) * 1000)
             failed_response = e.response
 
@@ -515,7 +515,7 @@ def _should_give_up(e: Exception) -> bool:
                     and 400 <= e.status_code < 500
                     and e.status_code != 429
                 )
-            if isinstance(e, httpx.HTTPStatusError):
+            if isinstance(e, httpx2.HTTPStatusError):
                 return (
                     e.response is not None
                     and e.response.status_code < 500

langfuse/_utils/request.py

@@ -4,7 +4,7 @@
 from base64 import b64encode
 from typing import Any, List, Union
 
-import httpx
+import httpx2
 
 from langfuse._utils.serializer import EventSerializer
 from langfuse.logger import langfuse_logger as logger
@@ -16,7 +16,7 @@ class LangfuseClient:
     _base_url: str
     _version: str
     _timeout: int
-    _session: httpx.Client
+    _session: httpx2.Client
 
     def __init__(
         self,
@@ -25,7 +25,7 @@ def __init__(
         base_url: str,
         version: str,
         timeout: int,
-        session: httpx.Client,
+        session: httpx2.Client,
     ):
         self._public_key = public_key
         self._secret_key = secret_key
@@ -46,15 +46,15 @@ def generate_headers(self) -> dict:
             "x-langfuse-public-key": self._public_key,
         }
 
-    def batch_post(self, **kwargs: Any) -> httpx.Response:
+    def batch_post(self, **kwargs: Any) -> httpx2.Response:
         """Post the `kwargs` to the batch API endpoint for events"""
         res = self.post(**kwargs)
 
         return self._process_response(
             res, success_message="data uploaded successfully", return_json=False
         )
 
-    def post(self, **kwargs: Any) -> httpx.Response:
+    def post(self, **kwargs: Any) -> httpx2.Response:
         """Post the `kwargs` to the API"""
         url = self._remove_trailing_slash(self._base_url) + "/api/public/ingestion"
         data = json.dumps(kwargs, cls=EventSerializer)
@@ -76,8 +76,8 @@ def _remove_trailing_slash(self, url: str) -> str:
         return url
 
     def _process_response(
-        self, res: httpx.Response, success_message: str, *, return_json: bool = True
-    ) -> Union[httpx.Response, Any]:
+        self, res: httpx2.Response, success_message: str, *, return_json: bool = True
+    ) -> Union[httpx2.Response, Any]:
         logger.debug("received response: %s", res.text)
         if res.status_code in (200, 201):
             logger.debug(success_message)

langfuse/api/client.py

@@ -4,7 +4,7 @@
 
 import typing
 
-import httpx
+import httpx2
 from .core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
 
 if typing.TYPE_CHECKING:
@@ -70,13 +70,13 @@ class LangfuseAPI:
         Additional headers to send with every request.
 
     timeout : typing.Optional[float]
-        The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx client is used, in which case this default is not enforced.
+        The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx2 client is used, in which case this default is not enforced.
 
     follow_redirects : typing.Optional[bool]
-        Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in.
+        Whether the default httpx2 client follows redirects or not, this is irrelevant if a custom httpx2 client is passed in.
 
-    httpx_client : typing.Optional[httpx.Client]
-        The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration.
+    httpx_client : typing.Optional[httpx2.Client]
+        The httpx2 client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx2 configuration.
 
     Examples
     --------
@@ -104,7 +104,7 @@ def __init__(
         headers: typing.Optional[typing.Dict[str, str]] = None,
         timeout: typing.Optional[float] = None,
         follow_redirects: typing.Optional[bool] = True,
-        httpx_client: typing.Optional[httpx.Client] = None,
+        httpx_client: typing.Optional[httpx2.Client] = None,
     ):
         _defaulted_timeout = (
             timeout
@@ -123,11 +123,11 @@ def __init__(
             headers=headers,
             httpx_client=httpx_client
             if httpx_client is not None
-            else httpx.Client(
+            else httpx2.Client(
                 timeout=_defaulted_timeout, follow_redirects=follow_redirects
             )
             if follow_redirects is not None
-            else httpx.Client(timeout=_defaulted_timeout),
+            else httpx2.Client(timeout=_defaulted_timeout),
             timeout=_defaulted_timeout,
         )
         self._annotation_queues: typing.Optional[AnnotationQueuesClient] = None
@@ -442,13 +442,13 @@ class AsyncLangfuseAPI:
         Additional headers to send with every request.
 
     timeout : typing.Optional[float]
-        The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx client is used, in which case this default is not enforced.
+        The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx2 client is used, in which case this default is not enforced.
 
     follow_redirects : typing.Optional[bool]
-        Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in.
+        Whether the default httpx2 client follows redirects or not, this is irrelevant if a custom httpx2 client is passed in.
 
-    httpx_client : typing.Optional[httpx.AsyncClient]
-        The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration.
+    httpx_client : typing.Optional[httpx2.AsyncClient]
+        The httpx2 client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx2 configuration.
 
     Examples
     --------
@@ -476,7 +476,7 @@ def __init__(
         headers: typing.Optional[typing.Dict[str, str]] = None,
         timeout: typing.Optional[float] = None,
         follow_redirects: typing.Optional[bool] = True,
-        httpx_client: typing.Optional[httpx.AsyncClient] = None,
+        httpx_client: typing.Optional[httpx2.AsyncClient] = None,
     ):
         _defaulted_timeout = (
             timeout
@@ -495,11 +495,11 @@ def __init__(
             headers=headers,
             httpx_client=httpx_client
             if httpx_client is not None
-            else httpx.AsyncClient(
+            else httpx2.AsyncClient(
                 timeout=_defaulted_timeout, follow_redirects=follow_redirects
             )
             if follow_redirects is not None
-            else httpx.AsyncClient(timeout=_defaulted_timeout),
+            else httpx2.AsyncClient(timeout=_defaulted_timeout),
             timeout=_defaulted_timeout,
         )
         self._annotation_queues: typing.Optional[AsyncAnnotationQueuesClient] = None

langfuse/api/core/client_wrapper.py

@@ -2,7 +2,7 @@
 
 import typing
 
-import httpx
+import httpx2
 from .http_client import AsyncHttpClient, HttpClient
 
 
@@ -36,7 +36,7 @@ def get_headers(self) -> typing.Dict[str, str]:
         username = self._get_username()
         password = self._get_password()
         if username is not None and password is not None:
-            headers["Authorization"] = httpx.BasicAuth(username, password)._auth_header
+            headers["Authorization"] = httpx2.BasicAuth(username, password)._auth_header
         if self._x_langfuse_sdk_name is not None:
             headers["X-Langfuse-Sdk-Name"] = self._x_langfuse_sdk_name
         if self._x_langfuse_sdk_version is not None:
@@ -79,7 +79,7 @@ def __init__(
         headers: typing.Optional[typing.Dict[str, str]] = None,
         base_url: str,
         timeout: typing.Optional[float] = None,
-        httpx_client: httpx.Client,
+        httpx_client: httpx2.Client,
     ):
         super().__init__(
             x_langfuse_sdk_name=x_langfuse_sdk_name,
@@ -112,7 +112,7 @@ def __init__(
         base_url: str,
         timeout: typing.Optional[float] = None,
         async_token: typing.Optional[typing.Callable[[], typing.Awaitable[str]]] = None,
-        httpx_client: httpx.AsyncClient,
+        httpx_client: httpx2.AsyncClient,
     ):
         super().__init__(
             x_langfuse_sdk_name=x_langfuse_sdk_name,