From d131f8d571187485f0cb42741d8303a0257a2e03 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sat, 22 Aug 2026 16:48:23 +0800 Subject: [PATCH] feat(network): migrate async HTTP client to HTTPX2 (#6394) --- app/adapters/external/market.py | 4 +- app/adapters/network/http.py | 107 +++++++++++++++--------------- app/modules/douban/apiv2.py | 6 +- docs/development-setup.md | 11 +++ pyproject.toml | 1 + tests/test_async_request_utils.py | 78 ++++++++++++++++++++-- tests/test_correlation.py | 7 +- tests/test_lifecycle_shutdown.py | 2 +- uv.lock | 47 +++++++++++++ 9 files changed, 198 insertions(+), 65 deletions(-) diff --git a/app/adapters/external/market.py b/app/adapters/external/market.py index 34c8ddd71..e39b5f010 100644 --- a/app/adapters/external/market.py +++ b/app/adapters/external/market.py @@ -20,7 +20,7 @@ from urllib.parse import parse_qs, quote, unquote, urlparse, urlsplit import aiofiles import aioshutil -import httpx +import httpx2 from anyio import Path as AsyncPath from packaging.markers import default_environment from packaging.requirements import Requirement @@ -2152,7 +2152,7 @@ class PluginHelper(metaclass=WeakSingleton): async def __async_request_with_fallback(url: str, headers: Optional[dict] = None, timeout: Optional[int] = 60, - is_api: bool = False) -> Optional[httpx.Response]: + is_api: bool = False) -> Optional[httpx2.Response]: """ 使用自动降级策略，异步请求资源，优先级依次为镜像站、代理、直连 :param url: 目标URL diff --git a/app/adapters/network/http.py b/app/adapters/network/http.py index 5fa000192..a0b6a40bf 100644 --- a/app/adapters/network/http.py +++ b/app/adapters/network/http.py @@ -10,7 +10,7 @@ from pathlib import Path from typing import Any, Dict, Optional, Tuple, Union import chardet -import httpx +import httpx2 import requests import urllib3 from requests import Response, Session @@ -31,7 +31,14 @@ def configure_default_user_agent(user_agent: str) -> None: _default_user_agent = user_agent -class _NonClosingTransportProxy(httpx.AsyncBaseTransport): +_ASYNC_STALE_CONNECTION_ERRORS = ( + httpx2.RemoteProtocolError, + httpx2.ReadError, + httpx2.WriteError, +) + + +class _NonClosingTransportProxy(httpx2.AsyncBaseTransport): """ 包装共享底层 transport，转发请求但吞掉 __aexit__/aclose 调用。 防止 per-call AsyncClient 在 async with 退出时把底层连接池一并清空。 @@ -40,7 +47,7 @@ class _NonClosingTransportProxy(httpx.AsyncBaseTransport): __slots__ = ("_wrapped",) - def __init__(self, wrapped: httpx.AsyncBaseTransport): + def __init__(self, wrapped: httpx2.AsyncBaseTransport): """保存由进程统一管理生命周期的底层传输对象。""" self._wrapped = wrapped @@ -58,7 +65,7 @@ class _NonClosingTransportProxy(httpx.AsyncBaseTransport): # 故意 no-op：调用方显式 aclose 也不影响共享池 return None - async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: """将异步请求转发给共享底层传输。""" return await self._wrapped.handle_async_request(request) @@ -73,7 +80,7 @@ _SharedTransportKey = Tuple[ ] # 共享底层 transport 桶，按事件循环和配置区分，支持 LRU 淘汰 -_shared_async_transports: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, collections.OrderedDict[_SharedTransportKey, httpx.AsyncHTTPTransport]] = weakref.WeakKeyDictionary() +_shared_async_transports: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, collections.OrderedDict[_SharedTransportKey, httpx2.AsyncHTTPTransport]] = weakref.WeakKeyDictionary() # 不同线程各自驱动的事件循环并发首次写入外层弱字典时，需要互斥保护 _shared_async_transports_lock = threading.Lock() # 每个事件循环允许的最大共享 transport 桶数；超出后按 LRU 淘汰最久未用桶。 @@ -87,7 +94,7 @@ _DEFAULT_KEEPALIVE_EXPIRY = 30 # 同步 requests.Session 复用连接时，遇到对端或代理关闭 keep-alive 后允许重试的方法 _REQUESTS_RETRY_IDEMPOTENT_METHODS = ("GET", "HEAD", "OPTIONS") -# 代理走 CONNECT 隧道时，httpx 默认开启的 HTTP/2 多路复用会把并发请求叠加到极少数隧道上； +# 代理走 CONNECT 隧道时，HTTP/2 多路复用会把并发请求叠加到极少数隧道上； # 隧道被代理节点切换或空闲回收打断后，复用其上的所有请求会同时失败。按 (proxy, host) 熔断： # 命中一次连接层失败就记录下次允许再尝试 h2 的时间戳（time.monotonic 基准），冷却期内该 # (proxy, host) 的请求直接退化为 http1.1；冷却期结束后自动恢复尝试 h2。 @@ -98,17 +105,17 @@ _h2_proxy_retry_at: Dict[Tuple[str, str], float] = {} # EndOfStream 等）；超时、连接失败、代理不可达等错误换 h1 一样会发生， # 不应触发熔断，也不值得付出一次注定同样失败的 h1 重试 _H2_TUNNEL_BREAK_ERRORS = ( - httpx.RemoteProtocolError, - httpx.LocalProtocolError, - httpx.ReadError, - httpx.WriteError, - httpx.CloseError, + httpx2.RemoteProtocolError, + httpx2.LocalProtocolError, + httpx2.ReadError, + httpx2.WriteError, + httpx2.CloseError, ) def _h2_proxy_breaker_key(proxy: str, url: str) -> Tuple[str, str]: try: - host = httpx.URL(url).host or "" + host = httpx2.URL(url).host or "" except Exception: host = url return proxy, host @@ -152,7 +159,7 @@ def _get_shared_async_transport( max_keepalive_connections: int, max_connections: int, keepalive_expiry: int, -) -> Optional[httpx.AsyncHTTPTransport]: +) -> Optional[httpx2.AsyncHTTPTransport]: """ 返回与当前事件循环绑定的共享 AsyncHTTPTransport（底层连接池）；首次按需创建。 没有运行中的事件循环或循环已关闭时返回 None，由调用方走临时客户端兜底。 @@ -161,7 +168,7 @@ def _get_shared_async_transport( 会话级状态由调用方在外层 AsyncClient(transport=...) 实例化时单独配置， 每次调用用完即销毁，因此天然无 jar 累积串扰。 """ - # 规范化代理：拒绝空字符串等非法值，防止 httpx 抛出 Unknown scheme for proxy URL + # 规范化代理：拒绝空字符串等非法值，防止客户端解析空代理地址失败 if proxy is not None and (not proxy or not proxy.strip()): proxy = None try: @@ -191,11 +198,11 @@ def _get_shared_async_transport( return transport # 首次见到这个配置，创建新的共享 transport 桶 - transport = httpx.AsyncHTTPTransport( + transport = httpx2.AsyncHTTPTransport( http2=http2, proxy=proxy, verify=verify, - limits=httpx.Limits( + limits=httpx2.Limits( max_keepalive_connections=max_keepalive_connections, max_connections=max_connections, keepalive_expiry=keepalive_expiry, @@ -978,7 +985,7 @@ class AsyncRequestUtils: ua: str = None, cookies: Union[str, dict] = None, proxies: dict = None, - client: httpx.AsyncClient = None, + client: httpx2.AsyncClient = None, timeout: int = None, referer: str = None, content_type: str = None, @@ -995,7 +1002,7 @@ class AsyncRequestUtils: :param ua: User-Agent字符串 :param cookies: Cookie字符串或字典 :param proxies: 代理设置 - :param client: httpx.AsyncClient实例，如果为None则创建新的客户端 + :param client: 调用方自管的 HTTPX2 AsyncClient；为空时使用宿主客户端 :param timeout: 请求超时时间，默认为20秒 :param referer: Referer头部信息 :param content_type: 请求的Content-Type，默认为 "application/x-www-form-urlencoded; charset=UTF-8" @@ -1096,7 +1103,7 @@ class AsyncRequestUtils: async def request( self, method: str, url: str, raise_exception: bool = False, **kwargs - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 发起异步HTTP请求 :param method: HTTP方法，如 get, post, put 等 @@ -1104,7 +1111,7 @@ class AsyncRequestUtils: :param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None :param kwargs: 其他请求参数，如headers, cookies, proxies等 :return: HTTP响应对象 - :raises: httpx.RequestError 仅raise_exception为True时会抛出 + :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出 """ # 运行时 self._cookies 只能是 dict | None（cookie_parse 默认 array=False 返回 dict） cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None @@ -1137,7 +1144,7 @@ class AsyncRequestUtils: return await self._dispatch_request( False, cookies_dict, method, url, raise_exception, **kwargs ) - except httpx.RequestError: + except httpx2.RequestError: # 与 h2 隧道无关的失败（超时、连接失败等）：不熔断也不重试， # 恢复调用方原本的 raise_exception 语义 if raise_exception: @@ -1147,7 +1154,7 @@ class AsyncRequestUtils: async def _dispatch_request( self, http2: bool, cookies_dict: Optional[dict], method: str, url: str, raise_exception: bool, **kwargs - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 按给定 http2 开关构建/复用底层连接并发起请求，供 request() 的 h2/h1 熔断切换复用 """ @@ -1165,9 +1172,9 @@ class AsyncRequestUtils: if transport is not None: # 用 _NonClosingTransportProxy 包装共享 transport，吞掉 AsyncClient.__aexit__ # 传播下来的 transport.__aexit__，避免每次 async with 退出都把共享连接池清空。 - async with httpx.AsyncClient( + async with httpx2.AsyncClient( transport=_NonClosingTransportProxy(transport), - timeout=httpx.Timeout(self._timeout), + timeout=httpx2.Timeout(self._timeout), follow_redirects=self._follow_redirects, cookies=cookies_dict, ) as client: @@ -1176,7 +1183,7 @@ class AsyncRequestUtils: ) # 兜底：没有运行中的事件循环时，临时客户端走完即关 - async with httpx.AsyncClient( + async with httpx2.AsyncClient( http2=http2, proxy=self._proxies, timeout=self._timeout, @@ -1190,12 +1197,12 @@ class AsyncRequestUtils: async def _make_request( self, - client: httpx.AsyncClient, + client: httpx2.AsyncClient, method: str, url: str, raise_exception: bool = False, **kwargs, - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 执行实际的异步请求 """ @@ -1213,15 +1220,13 @@ class AsyncRequestUtils: # 仅对幂等方法做 stale-pool 竞态重试：复用了刚被对端 FIN 的 keep-alive 连接时， # 实际请求通常未到服务端，httpx 自身不重试，这里兜底一次。 is_idempotent = method_upper in ("GET", "HEAD", "OPTIONS") - stale_conn_errs = (httpx.RemoteProtocolError, httpx.ReadError, httpx.WriteError) - try: return await client.request(method, url, **kwargs) - except stale_conn_errs: + except _ASYNC_STALE_CONNECTION_ERRORS: if is_idempotent: try: return await client.request(method, url, **kwargs) - except httpx.RequestError: + except httpx2.RequestError: if raise_exception: raise return None @@ -1230,7 +1235,7 @@ class AsyncRequestUtils: if raise_exception: raise return None - except httpx.RequestError: + except httpx2.RequestError: if raise_exception: raise return None @@ -1258,7 +1263,7 @@ class AsyncRequestUtils: async def post( self, url: str, data: Any = None, json: dict = None, **kwargs - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 发送异步POST请求 :param url: 请求的URL @@ -1273,7 +1278,7 @@ class AsyncRequestUtils: async def put( self, url: str, data: Any = None, **kwargs - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 发送异步PUT请求 :param url: 请求的URL @@ -1292,7 +1297,7 @@ class AsyncRequestUtils: allow_redirects: bool = True, raise_exception: bool = False, **kwargs, - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 发送异步GET请求并返回响应对象 :param url: 请求的URL @@ -1303,7 +1308,7 @@ class AsyncRequestUtils: :param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None :param kwargs: 其他请求参数，如headers, cookies, proxies等 :return: HTTP响应对象，若发生RequestError则返回None - :raises: httpx.RequestError 仅raise_exception为True时会抛出 + :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出 """ return await self.request( method="get", @@ -1326,13 +1331,13 @@ class AsyncRequestUtils: ): """ 获取异步流式响应的上下文管理器，适用于大文件下载。 - 使用 httpx.AsyncClient.stream() 标准流式 API，避免把响应体一次性读入内存。 + 使用 AsyncClient.stream() 标准流式 API，避免把响应体一次性读入内存。 :param url: 请求的URL :param params: 请求的参数 :param raise_exception: 是否在发生异常时抛出，否则吞掉并 yield None :param kwargs: 其他请求参数（headers, cookies 等） - :return: 上下文管理器，进入后 yield httpx.Response（出错时 yield None） + :return: 上下文管理器，进入后返回响应对象（出错时返回 None） """ cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None kwargs["headers"] = with_correlation_header( @@ -1341,8 +1346,6 @@ class AsyncRequestUtils: # 与 _make_request 保持一致：复用 keep-alive 时偶遇对端 FIN 的连接， # 流式 GET 是幂等的，单次重试即可 - stale_conn_errs = (httpx.RemoteProtocolError, httpx.ReadError, httpx.WriteError) - async with AsyncExitStack() as stack: # 选 client：复用与 request() 相同的三条 path 逻辑 if self._client is not None: @@ -1360,16 +1363,16 @@ class AsyncRequestUtils: ) if transport is not None: client = await stack.enter_async_context( - httpx.AsyncClient( + httpx2.AsyncClient( transport=_NonClosingTransportProxy(transport), - timeout=httpx.Timeout(self._timeout), + timeout=httpx2.Timeout(self._timeout), follow_redirects=self._follow_redirects, cookies=cookies_dict, ) ) else: client = await stack.enter_async_context( - httpx.AsyncClient( + httpx2.AsyncClient( http2=self._http2, proxy=self._proxies, timeout=self._timeout, @@ -1383,17 +1386,17 @@ class AsyncRequestUtils: response = await stack.enter_async_context( client.stream("GET", url, params=params, **kwargs) ) - except stale_conn_errs: + except _ASYNC_STALE_CONNECTION_ERRORS: try: response = await stack.enter_async_context( client.stream("GET", url, params=params, **kwargs) ) - except httpx.RequestError: + except httpx2.RequestError: if raise_exception: raise yield None return - except httpx.RequestError: + except httpx2.RequestError: if raise_exception: raise yield None @@ -1413,7 +1416,7 @@ class AsyncRequestUtils: json: dict = None, raise_exception: bool = False, **kwargs, - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 发送异步POST请求并返回响应对象 :param url: 请求的URL @@ -1425,7 +1428,7 @@ class AsyncRequestUtils: :param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None :param kwargs: 其他请求参数，如headers, cookies, proxies等 :return: HTTP响应对象，若发生RequestError则返回None - :raises: httpx.RequestError 仅raise_exception为True时会抛出 + :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出 """ return await self.request( method="post", @@ -1449,7 +1452,7 @@ class AsyncRequestUtils: json: dict = None, raise_exception: bool = False, **kwargs, - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 发送异步PUT请求并返回响应对象 :param url: 请求的URL @@ -1461,7 +1464,7 @@ class AsyncRequestUtils: :param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None :param kwargs: 其他请求参数，如headers, cookies, proxies等 :return: HTTP响应对象，若发生RequestError则返回None - :raises: httpx.RequestError 仅raise_exception为True时会抛出 + :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出 """ return await self.request( method="put", @@ -1483,7 +1486,7 @@ class AsyncRequestUtils: allow_redirects: bool = True, raise_exception: bool = False, **kwargs, - ) -> Optional[httpx.Response]: + ) -> Optional[httpx2.Response]: """ 发送异步DELETE请求并返回响应对象 :param url: 请求的URL @@ -1493,7 +1496,7 @@ class AsyncRequestUtils: :param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None :param kwargs: 其他请求参数，如headers, cookies, proxies等 :return: HTTP响应对象，若发生RequestError则返回None - :raises: httpx.RequestError 仅raise_exception为True时会抛出 + :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出 """ return await self.request( method="delete", diff --git a/app/modules/douban/apiv2.py b/app/modules/douban/apiv2.py index 77d605aac..49ddeb46d 100644 --- a/app/modules/douban/apiv2.py +++ b/app/modules/douban/apiv2.py @@ -8,7 +8,7 @@ from random import choice from typing import Optional, Union from urllib import parse -import httpx +import httpx2 import requests from bs4 import BeautifulSoup @@ -223,7 +223,9 @@ class DoubanApi(metaclass=WeakSingleton): return req_url, params @staticmethod - def _handle_response(resp: Union[requests.Response, httpx.Response]) -> dict: + def _handle_response( + resp: Union[requests.Response, httpx2.Response] + ) -> dict: """ 处理HTTP响应 """ diff --git a/docs/development-setup.md b/docs/development-setup.md index f1db3c475..665e02299 100644 --- a/docs/development-setup.md +++ b/docs/development-setup.md @@ -122,6 +122,17 @@ dependencies = ["example-package>=1,<2"] - 仅有 `requirements.txt` 的历史插件继续按原方式安装； - 宿主不消费插件自己的 `uv.lock`，因为多个插件共享同一主程序环境，不能分别同步独立锁文件。 +### 3.2 异步 HTTP 客户端边界 + +主程序自建的 `AsyncRequestUtils` 使用 HTTPX2，`app.sdk.network.AsyncRequestUtils` 与旧插件 +入口 `app.utils.http.AsyncRequestUtils` 共享同一实现。未显式传入客户端时，返回的响应与抛出的 +请求异常均来自 `httpx2`；直接依赖响应类型或异常类型的 V3 代码应导入 `httpx2`。 + +OpenAI、Anthropic、Google GenAI、LangChain、CloakBrowser 等第三方 SDK 继续使用它们声明的 +HTTPX 版本。不得调用 `httpx2.alias_httpx()` 在进程内替换 `httpx`，否则会同时改变第三方 SDK、 +测试工具和插件的导入结果。确需复用调用方自管客户端时，向 `AsyncRequestUtils` 传入 +`httpx2.AsyncClient`。 + ### 4. 准备资源与插件目录 本地源码开发时，主程序需要读取资源文件和插件源码。相关文件需要放到主程序实际加载的目录下： diff --git a/pyproject.toml b/pyproject.toml index b20637863..9ff528e6a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ dependencies = [ "fastapi~=0.141.1", "google-genai~=2.8.0", "httpx[http2,socks]~=0.28.1", + "httpx2[http2,socks]~=2.12.0", "jieba-next~=1.0.0rc1", "jinja2~=3.1.6", "langchain~=1.3.15", diff --git a/tests/test_async_request_utils.py b/tests/test_async_request_utils.py index 841fed84a..1d3a03c15 100644 --- a/tests/test_async_request_utils.py +++ b/tests/test_async_request_utils.py @@ -1,11 +1,12 @@ import asyncio import time -import httpx +import httpx2 import pytest from app.adapters.network import http as http_module from app.adapters.network.http import AsyncRequestUtils +from app.sdk.network import AsyncRequestUtils as SdkAsyncRequestUtils PROXY = "http://proxy.example:7890" URL = "https://raw.githubusercontent.com/demo/repo/main/package.json" @@ -24,7 +25,7 @@ def _fake_dispatch(calls, fail_when): calls.append(http2) if fail_when(http2): if raise_exception: - raise httpx.RemoteProtocolError("tunnel closed") + raise httpx2.RemoteProtocolError("tunnel closed") return None return "ok" @@ -92,7 +93,7 @@ def test_timeout_does_not_trip_breaker_or_retry(monkeypatch): async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs): calls.append(http2) - raise httpx.ConnectTimeout("proxy slow") + raise httpx2.ConnectTimeout("proxy slow") monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake) @@ -112,12 +113,12 @@ def test_timeout_still_raises_when_raise_exception_enabled(monkeypatch): async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs): calls.append(http2) - raise httpx.ConnectTimeout("proxy slow") + raise httpx2.ConnectTimeout("proxy slow") monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake) utils = AsyncRequestUtils(proxies={"https": PROXY}) - with pytest.raises(httpx.ConnectTimeout): + with pytest.raises(httpx2.ConnectTimeout): asyncio.run(utils.request("get", URL, raise_exception=True)) assert calls == [True] @@ -154,3 +155,70 @@ def test_no_proxy_configured_skips_breaker_logic(monkeypatch): assert result == "ok" assert calls == [True] + + +def test_internal_client_uses_httpx2_transport(monkeypatch): + """宿主自建客户端使用 HTTPX2，并返回对应响应对象。""" + + async def respond(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, json={"client": "httpx2"}, request=request) + + monkeypatch.setattr( + http_module, + "_get_shared_async_transport", + lambda **_kwargs: httpx2.MockTransport(respond), + ) + + response = asyncio.run( + AsyncRequestUtils().get_res("https://example.com/data", raise_exception=True) + ) + + assert isinstance(response, httpx2.Response) + assert response.json() == {"client": "httpx2"} + + +def test_plugin_sdk_uses_httpx2_by_default(monkeypatch): + """插件 SDK 默认复用宿主 HTTPX2 客户端合同。""" + + async def respond(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, request=request) + + monkeypatch.setattr( + http_module, + "_get_shared_async_transport", + lambda **_kwargs: httpx2.MockTransport(respond), + ) + + response = asyncio.run( + SdkAsyncRequestUtils().get_res( + "https://example.com/sdk", raise_exception=True + ) + ) + + assert SdkAsyncRequestUtils is AsyncRequestUtils + assert isinstance(response, httpx2.Response) + + +def test_legacy_http_module_uses_httpx2_by_default(monkeypatch): + """旧插件 HTTP 导入入口随宿主默认客户端迁移到 HTTPX2。""" + import importlib + + legacy_http = importlib.import_module("app.utils.http") + + async def respond(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, request=request) + + monkeypatch.setattr( + http_module, + "_get_shared_async_transport", + lambda **_kwargs: httpx2.MockTransport(respond), + ) + + response = asyncio.run( + legacy_http.AsyncRequestUtils().get_res( + "https://example.com/legacy", raise_exception=True + ) + ) + + assert legacy_http.AsyncRequestUtils is AsyncRequestUtils + assert isinstance(response, httpx2.Response) diff --git a/tests/test_correlation.py b/tests/test_correlation.py index e38a9be18..ee211a6ff 100644 --- a/tests/test_correlation.py +++ b/tests/test_correlation.py @@ -6,6 +6,7 @@ from types import SimpleNamespace from unittest.mock import MagicMock import httpx +import httpx2 import pytest from starlette.applications import Starlette from starlette.responses import JSONResponse, StreamingResponse @@ -162,12 +163,12 @@ async def test_async_external_request_preserves_explicit_header() -> None: """异步外呼默认传播当前 ID，但不得覆盖调用方显式 trace 边界。""" observed = [] - async def respond(request: httpx.Request) -> httpx.Response: + async def respond(request: httpx2.Request) -> httpx2.Response: """记录 MockTransport 收到的请求头。""" observed.append(request.headers[CORRELATION_ID_HEADER]) - return httpx.Response(200) + return httpx2.Response(200) - async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + async with httpx2.AsyncClient(transport=httpx2.MockTransport(respond)) as client: utils = AsyncRequestUtils(client=client) with correlation_scope("context-request"): await utils.request("GET", "https://example.com/default") diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index c8c3a03ab..f287a4438 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -658,7 +658,7 @@ def test_shared_http_close_waits_for_real_lru_eviction(monkeypatch): raise RuntimeError("eviction close failed") monkeypatch.setattr(http_utils, "_MAX_SHARED_TRANSPORTS_PER_LOOP", 1) - monkeypatch.setattr(http_utils.httpx, "AsyncHTTPTransport", FakeTransport) + monkeypatch.setattr(http_utils.httpx2, "AsyncHTTPTransport", FakeTransport) async def run_test(): transport_kwargs = { "proxy": None, diff --git a/uv.lock b/uv.lock index ba6ed2875..e5c860ae0 100644 --- a/uv.lock +++ b/uv.lock @@ -967,6 +967,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, ] +[[package]] +name = "httpcore2" +version = "2.12.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "h11" }, + { name = "truststore" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/be/ad/f4f0e57345f1870f3e8cb624e058d7eca6e5a27d33bcc3311d9b618734cd/httpcore2-2.12.0.tar.gz", hash = "sha256:9293522bba0aa7c4c8e9e3f040c16575bd8868e155a77fa30c7a9085a5eae648", size = 67548, upload-time = "2026-08-18T13:22:08.211Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d2/74/d370e55600d9bcfa0d9794b0166126d49291a3d2b20c268fc98c453a4948/httpcore2-2.12.0-py3-none-any.whl", hash = "sha256:7e04258ce01013d7d615e5b910a3b27fac937d7a95038227e79652b4ba3b4ceb", size = 83074, upload-time = "2026-08-18T13:22:05.854Z" }, +] + [[package]] name = "httpx" version = "0.28.1" @@ -1003,6 +1016,29 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d2/fd/6668e5aec43ab844de6fc74927e155a3b37bf40d7c3790e49fc0406b6578/httpx_sse-0.4.3-py3-none-any.whl", hash = "sha256:0ac1c9fe3c0afad2e0ebb25a934a59f4c7823b60792691f779fad2c5568830fc", size = 8960, upload-time = "2025-10-10T21:48:21.158Z" }, ] +[[package]] +name = "httpx2" +version = "2.12.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "httpcore2" }, + { name = "idna" }, + { name = "truststore" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7f/f8/579a8b51e42e38ee32647df9f08aa25643ae788e275cc625b199829c4671/httpx2-2.12.0.tar.gz", hash = "sha256:7631fe9887a8a2275f4a2540e053aa670fcc50742864a9ae7c66e609fdcf12cf", size = 100040, upload-time = "2026-08-18T13:22:09.086Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c8/95/411ba65569158e862368917aaf56597f3e5fa3b91b0502919638465a08f3/httpx2-2.12.0-py3-none-any.whl", hash = "sha256:cc8b6eecb8661c146b8f89a60e97456ee086e91a784ed31ac450c3a9e613dd36", size = 95427, upload-time = "2026-08-18T13:22:06.834Z" }, +] + +[package.optional-dependencies] +http2 = [ + { name = "h2" }, +] +socks = [ + { name = "socksio" }, +] + [[package]] name = "hyperframe" version = "6.1.0" @@ -1500,6 +1536,7 @@ dependencies = [ { name = "fastapi" }, { name = "google-genai" }, { name = "httpx", extra = ["http2", "socks"] }, + { name = "httpx2", extra = ["http2", "socks"] }, { name = "jieba-next" }, { name = "jinja2" }, { name = "langchain" }, @@ -1609,6 +1646,7 @@ requires-dist = [ { name = "fastapi", specifier = "~=0.141.1" }, { name = "google-genai", specifier = "~=2.8.0" }, { name = "httpx", extras = ["http2", "socks"], specifier = "~=0.28.1" }, + { name = "httpx2", extras = ["http2", "socks"], specifier = "~=2.12.0" }, { name = "jieba-next", specifier = "~=1.0.0rc1" }, { name = "jinja2", specifier = "~=3.1.6" }, { name = "langchain", specifier = "~=1.3.15" }, @@ -2977,6 +3015,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/80/9c/edc8ee0d9ab7a05f0cb334fc506ce3749ff869a8d41683108313096ce5ab/transmission_rpc-7.0.12-py3-none-any.whl", hash = "sha256:85e26b7ca13e1e102a695fe19f19f4236bd42bb4920b0edb9e35f6bb46bbe86e", size = 29103, upload-time = "2026-08-09T14:37:40.272Z" }, ] +[[package]] +name = "truststore" +version = "0.10.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/53/a3/1585216310e344e8102c22482f6060c7a6ea0322b63e026372e6dcefcfd6/truststore-0.10.4.tar.gz", hash = "sha256:9d91bd436463ad5e4ee4aba766628dd6cd7010cf3e2461756b3303710eebc301", size = 26169, upload-time = "2025-08-12T18:49:02.73Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.46Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0"