feat(network): migrate async HTTP client to HTTPX2 (#6394)

This commit is contained in:
InfinityPacer
2026-08-22 16:48:23 +08:00
committed by GitHub
parent dfc047b880
commit d131f8d571
9 changed files with 198 additions and 65 deletions
+2 -2
View File
@@ -20,7 +20,7 @@ from urllib.parse import parse_qs, quote, unquote, urlparse, urlsplit
import aiofiles import aiofiles
import aioshutil import aioshutil
import httpx import httpx2
from anyio import Path as AsyncPath from anyio import Path as AsyncPath
from packaging.markers import default_environment from packaging.markers import default_environment
from packaging.requirements import Requirement from packaging.requirements import Requirement
@@ -2152,7 +2152,7 @@ class PluginHelper(metaclass=WeakSingleton):
async def __async_request_with_fallback(url: str, async def __async_request_with_fallback(url: str,
headers: Optional[dict] = None, headers: Optional[dict] = None,
timeout: Optional[int] = 60, timeout: Optional[int] = 60,
is_api: bool = False) -> Optional[httpx.Response]: is_api: bool = False) -> Optional[httpx2.Response]:
""" """
使用自动降级策略,异步请求资源,优先级依次为镜像站、代理、直连 使用自动降级策略,异步请求资源,优先级依次为镜像站、代理、直连
:param url: 目标URL :param url: 目标URL
+55 -52
View File
@@ -10,7 +10,7 @@ from pathlib import Path
from typing import Any, Dict, Optional, Tuple, Union from typing import Any, Dict, Optional, Tuple, Union
import chardet import chardet
import httpx import httpx2
import requests import requests
import urllib3 import urllib3
from requests import Response, Session from requests import Response, Session
@@ -31,7 +31,14 @@ def configure_default_user_agent(user_agent: str) -> None:
_default_user_agent = user_agent _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 调用。 包装共享底层 transport,转发请求但吞掉 __aexit__/aclose 调用。
防止 per-call AsyncClient 在 async with 退出时把底层连接池一并清空。 防止 per-call AsyncClient 在 async with 退出时把底层连接池一并清空。
@@ -40,7 +47,7 @@ class _NonClosingTransportProxy(httpx.AsyncBaseTransport):
__slots__ = ("_wrapped",) __slots__ = ("_wrapped",)
def __init__(self, wrapped: httpx.AsyncBaseTransport): def __init__(self, wrapped: httpx2.AsyncBaseTransport):
"""保存由进程统一管理生命周期的底层传输对象。""" """保存由进程统一管理生命周期的底层传输对象。"""
self._wrapped = wrapped self._wrapped = wrapped
@@ -58,7 +65,7 @@ class _NonClosingTransportProxy(httpx.AsyncBaseTransport):
# 故意 no-op:调用方显式 aclose 也不影响共享池 # 故意 no-op:调用方显式 aclose 也不影响共享池
return None 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) return await self._wrapped.handle_async_request(request)
@@ -73,7 +80,7 @@ _SharedTransportKey = Tuple[
] ]
# 共享底层 transport 桶,按事件循环和配置区分,支持 LRU 淘汰 # 共享底层 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() _shared_async_transports_lock = threading.Lock()
# 每个事件循环允许的最大共享 transport 桶数;超出后按 LRU 淘汰最久未用桶。 # 每个事件循环允许的最大共享 transport 桶数;超出后按 LRU 淘汰最久未用桶。
@@ -87,7 +94,7 @@ _DEFAULT_KEEPALIVE_EXPIRY = 30
# 同步 requests.Session 复用连接时,遇到对端或代理关闭 keep-alive 后允许重试的方法 # 同步 requests.Session 复用连接时,遇到对端或代理关闭 keep-alive 后允许重试的方法
_REQUESTS_RETRY_IDEMPOTENT_METHODS = ("GET", "HEAD", "OPTIONS") _REQUESTS_RETRY_IDEMPOTENT_METHODS = ("GET", "HEAD", "OPTIONS")
# 代理走 CONNECT 隧道时,httpx 默认开启的 HTTP/2 多路复用会把并发请求叠加到极少数隧道上; # 代理走 CONNECT 隧道时,HTTP/2 多路复用会把并发请求叠加到极少数隧道上;
# 隧道被代理节点切换或空闲回收打断后,复用其上的所有请求会同时失败。按 (proxy, host) 熔断: # 隧道被代理节点切换或空闲回收打断后,复用其上的所有请求会同时失败。按 (proxy, host) 熔断:
# 命中一次连接层失败就记录下次允许再尝试 h2 的时间戳(time.monotonic 基准),冷却期内该 # 命中一次连接层失败就记录下次允许再尝试 h2 的时间戳(time.monotonic 基准),冷却期内该
# (proxy, host) 的请求直接退化为 http1.1;冷却期结束后自动恢复尝试 h2。 # (proxy, host) 的请求直接退化为 http1.1;冷却期结束后自动恢复尝试 h2。
@@ -98,17 +105,17 @@ _h2_proxy_retry_at: Dict[Tuple[str, str], float] = {}
# EndOfStream 等);超时、连接失败、代理不可达等错误换 h1 一样会发生, # EndOfStream 等);超时、连接失败、代理不可达等错误换 h1 一样会发生,
# 不应触发熔断,也不值得付出一次注定同样失败的 h1 重试 # 不应触发熔断,也不值得付出一次注定同样失败的 h1 重试
_H2_TUNNEL_BREAK_ERRORS = ( _H2_TUNNEL_BREAK_ERRORS = (
httpx.RemoteProtocolError, httpx2.RemoteProtocolError,
httpx.LocalProtocolError, httpx2.LocalProtocolError,
httpx.ReadError, httpx2.ReadError,
httpx.WriteError, httpx2.WriteError,
httpx.CloseError, httpx2.CloseError,
) )
def _h2_proxy_breaker_key(proxy: str, url: str) -> Tuple[str, str]: def _h2_proxy_breaker_key(proxy: str, url: str) -> Tuple[str, str]:
try: try:
host = httpx.URL(url).host or "" host = httpx2.URL(url).host or ""
except Exception: except Exception:
host = url host = url
return proxy, host return proxy, host
@@ -152,7 +159,7 @@ def _get_shared_async_transport(
max_keepalive_connections: int, max_keepalive_connections: int,
max_connections: int, max_connections: int,
keepalive_expiry: int, keepalive_expiry: int,
) -> Optional[httpx.AsyncHTTPTransport]: ) -> Optional[httpx2.AsyncHTTPTransport]:
""" """
返回与当前事件循环绑定的共享 AsyncHTTPTransport(底层连接池);首次按需创建。 返回与当前事件循环绑定的共享 AsyncHTTPTransport(底层连接池);首次按需创建。
没有运行中的事件循环或循环已关闭时返回 None,由调用方走临时客户端兜底。 没有运行中的事件循环或循环已关闭时返回 None,由调用方走临时客户端兜底。
@@ -161,7 +168,7 @@ def _get_shared_async_transport(
会话级状态由调用方在外层 AsyncClient(transport=...) 实例化时单独配置, 会话级状态由调用方在外层 AsyncClient(transport=...) 实例化时单独配置,
每次调用用完即销毁,因此天然无 jar 累积串扰。 每次调用用完即销毁,因此天然无 jar 累积串扰。
""" """
# 规范化代理:拒绝空字符串等非法值,防止 httpx 抛出 Unknown scheme for proxy URL # 规范化代理:拒绝空字符串等非法值,防止客户端解析空代理地址失败
if proxy is not None and (not proxy or not proxy.strip()): if proxy is not None and (not proxy or not proxy.strip()):
proxy = None proxy = None
try: try:
@@ -191,11 +198,11 @@ def _get_shared_async_transport(
return transport return transport
# 首次见到这个配置,创建新的共享 transport 桶 # 首次见到这个配置,创建新的共享 transport 桶
transport = httpx.AsyncHTTPTransport( transport = httpx2.AsyncHTTPTransport(
http2=http2, http2=http2,
proxy=proxy, proxy=proxy,
verify=verify, verify=verify,
limits=httpx.Limits( limits=httpx2.Limits(
max_keepalive_connections=max_keepalive_connections, max_keepalive_connections=max_keepalive_connections,
max_connections=max_connections, max_connections=max_connections,
keepalive_expiry=keepalive_expiry, keepalive_expiry=keepalive_expiry,
@@ -978,7 +985,7 @@ class AsyncRequestUtils:
ua: str = None, ua: str = None,
cookies: Union[str, dict] = None, cookies: Union[str, dict] = None,
proxies: dict = None, proxies: dict = None,
client: httpx.AsyncClient = None, client: httpx2.AsyncClient = None,
timeout: int = None, timeout: int = None,
referer: str = None, referer: str = None,
content_type: str = None, content_type: str = None,
@@ -995,7 +1002,7 @@ class AsyncRequestUtils:
:param ua: User-Agent字符串 :param ua: User-Agent字符串
:param cookies: Cookie字符串或字典 :param cookies: Cookie字符串或字典
:param proxies: 代理设置 :param proxies: 代理设置
:param client: httpx.AsyncClient实例,如果为None则创建新的客户端 :param client: 调用方自管的 HTTPX2 AsyncClient;为空时使用宿主客户端
:param timeout: 请求超时时间,默认为20秒 :param timeout: 请求超时时间,默认为20秒
:param referer: Referer头部信息 :param referer: Referer头部信息
:param content_type: 请求的Content-Type,默认为 "application/x-www-form-urlencoded; charset=UTF-8" :param content_type: 请求的Content-Type,默认为 "application/x-www-form-urlencoded; charset=UTF-8"
@@ -1096,7 +1103,7 @@ class AsyncRequestUtils:
async def request( async def request(
self, method: str, url: str, raise_exception: bool = False, **kwargs self, method: str, url: str, raise_exception: bool = False, **kwargs
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
发起异步HTTP请求 发起异步HTTP请求
:param method: HTTP方法,如 get, post, put 等 :param method: HTTP方法,如 get, post, put 等
@@ -1104,7 +1111,7 @@ class AsyncRequestUtils:
:param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None :param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None
:param kwargs: 其他请求参数,如headers, cookies, proxies等 :param kwargs: 其他请求参数,如headers, cookies, proxies等
:return: HTTP响应对象 :return: HTTP响应对象
:raises: httpx.RequestError 仅raise_exceptionTrue时抛出 :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出
""" """
# 运行时 self._cookies 只能是 dict | Nonecookie_parse 默认 array=False 返回 dict # 运行时 self._cookies 只能是 dict | Nonecookie_parse 默认 array=False 返回 dict
cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None
@@ -1137,7 +1144,7 @@ class AsyncRequestUtils:
return await self._dispatch_request( return await self._dispatch_request(
False, cookies_dict, method, url, raise_exception, **kwargs False, cookies_dict, method, url, raise_exception, **kwargs
) )
except httpx.RequestError: except httpx2.RequestError:
# 与 h2 隧道无关的失败(超时、连接失败等):不熔断也不重试, # 与 h2 隧道无关的失败(超时、连接失败等):不熔断也不重试,
# 恢复调用方原本的 raise_exception 语义 # 恢复调用方原本的 raise_exception 语义
if raise_exception: if raise_exception:
@@ -1147,7 +1154,7 @@ class AsyncRequestUtils:
async def _dispatch_request( async def _dispatch_request(
self, http2: bool, cookies_dict: Optional[dict], method: str, url: str, self, http2: bool, cookies_dict: Optional[dict], method: str, url: str,
raise_exception: bool, **kwargs raise_exception: bool, **kwargs
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
按给定 http2 开关构建/复用底层连接并发起请求,供 request() 的 h2/h1 熔断切换复用 按给定 http2 开关构建/复用底层连接并发起请求,供 request() 的 h2/h1 熔断切换复用
""" """
@@ -1165,9 +1172,9 @@ class AsyncRequestUtils:
if transport is not None: if transport is not None:
# 用 _NonClosingTransportProxy 包装共享 transport,吞掉 AsyncClient.__aexit__ # 用 _NonClosingTransportProxy 包装共享 transport,吞掉 AsyncClient.__aexit__
# 传播下来的 transport.__aexit__,避免每次 async with 退出都把共享连接池清空。 # 传播下来的 transport.__aexit__,避免每次 async with 退出都把共享连接池清空。
async with httpx.AsyncClient( async with httpx2.AsyncClient(
transport=_NonClosingTransportProxy(transport), transport=_NonClosingTransportProxy(transport),
timeout=httpx.Timeout(self._timeout), timeout=httpx2.Timeout(self._timeout),
follow_redirects=self._follow_redirects, follow_redirects=self._follow_redirects,
cookies=cookies_dict, cookies=cookies_dict,
) as client: ) as client:
@@ -1176,7 +1183,7 @@ class AsyncRequestUtils:
) )
# 兜底:没有运行中的事件循环时,临时客户端走完即关 # 兜底:没有运行中的事件循环时,临时客户端走完即关
async with httpx.AsyncClient( async with httpx2.AsyncClient(
http2=http2, http2=http2,
proxy=self._proxies, proxy=self._proxies,
timeout=self._timeout, timeout=self._timeout,
@@ -1190,12 +1197,12 @@ class AsyncRequestUtils:
async def _make_request( async def _make_request(
self, self,
client: httpx.AsyncClient, client: httpx2.AsyncClient,
method: str, method: str,
url: str, url: str,
raise_exception: bool = False, raise_exception: bool = False,
**kwargs, **kwargs,
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
执行实际的异步请求 执行实际的异步请求
""" """
@@ -1213,15 +1220,13 @@ class AsyncRequestUtils:
# 仅对幂等方法做 stale-pool 竞态重试:复用了刚被对端 FIN 的 keep-alive 连接时, # 仅对幂等方法做 stale-pool 竞态重试:复用了刚被对端 FIN 的 keep-alive 连接时,
# 实际请求通常未到服务端,httpx 自身不重试,这里兜底一次。 # 实际请求通常未到服务端,httpx 自身不重试,这里兜底一次。
is_idempotent = method_upper in ("GET", "HEAD", "OPTIONS") is_idempotent = method_upper in ("GET", "HEAD", "OPTIONS")
stale_conn_errs = (httpx.RemoteProtocolError, httpx.ReadError, httpx.WriteError)
try: try:
return await client.request(method, url, **kwargs) return await client.request(method, url, **kwargs)
except stale_conn_errs: except _ASYNC_STALE_CONNECTION_ERRORS:
if is_idempotent: if is_idempotent:
try: try:
return await client.request(method, url, **kwargs) return await client.request(method, url, **kwargs)
except httpx.RequestError: except httpx2.RequestError:
if raise_exception: if raise_exception:
raise raise
return None return None
@@ -1230,7 +1235,7 @@ class AsyncRequestUtils:
if raise_exception: if raise_exception:
raise raise
return None return None
except httpx.RequestError: except httpx2.RequestError:
if raise_exception: if raise_exception:
raise raise
return None return None
@@ -1258,7 +1263,7 @@ class AsyncRequestUtils:
async def post( async def post(
self, url: str, data: Any = None, json: dict = None, **kwargs self, url: str, data: Any = None, json: dict = None, **kwargs
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
发送异步POST请求 发送异步POST请求
:param url: 请求的URL :param url: 请求的URL
@@ -1273,7 +1278,7 @@ class AsyncRequestUtils:
async def put( async def put(
self, url: str, data: Any = None, **kwargs self, url: str, data: Any = None, **kwargs
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
发送异步PUT请求 发送异步PUT请求
:param url: 请求的URL :param url: 请求的URL
@@ -1292,7 +1297,7 @@ class AsyncRequestUtils:
allow_redirects: bool = True, allow_redirects: bool = True,
raise_exception: bool = False, raise_exception: bool = False,
**kwargs, **kwargs,
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
发送异步GET请求并返回响应对象 发送异步GET请求并返回响应对象
:param url: 请求的URL :param url: 请求的URL
@@ -1303,7 +1308,7 @@ class AsyncRequestUtils:
:param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None :param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None
:param kwargs: 其他请求参数,如headers, cookies, proxies等 :param kwargs: 其他请求参数,如headers, cookies, proxies等
:return: HTTP响应对象,若发生RequestError则返回None :return: HTTP响应对象,若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exceptionTrue时抛出 :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出
""" """
return await self.request( return await self.request(
method="get", method="get",
@@ -1326,13 +1331,13 @@ class AsyncRequestUtils:
): ):
""" """
获取异步流式响应的上下文管理器,适用于大文件下载。 获取异步流式响应的上下文管理器,适用于大文件下载。
使用 httpx.AsyncClient.stream() 标准流式 API,避免把响应体一次性读入内存。 使用 AsyncClient.stream() 标准流式 API,避免把响应体一次性读入内存。
:param url: 请求的URL :param url: 请求的URL
:param params: 请求的参数 :param params: 请求的参数
:param raise_exception: 是否在发生异常时抛出,否则吞掉并 yield None :param raise_exception: 是否在发生异常时抛出,否则吞掉并 yield None
:param kwargs: 其他请求参数(headers, cookies 等) :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 cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None
kwargs["headers"] = with_correlation_header( kwargs["headers"] = with_correlation_header(
@@ -1341,8 +1346,6 @@ class AsyncRequestUtils:
# 与 _make_request 保持一致:复用 keep-alive 时偶遇对端 FIN 的连接, # 与 _make_request 保持一致:复用 keep-alive 时偶遇对端 FIN 的连接,
# 流式 GET 是幂等的,单次重试即可 # 流式 GET 是幂等的,单次重试即可
stale_conn_errs = (httpx.RemoteProtocolError, httpx.ReadError, httpx.WriteError)
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
# 选 client:复用与 request() 相同的三条 path 逻辑 # 选 client:复用与 request() 相同的三条 path 逻辑
if self._client is not None: if self._client is not None:
@@ -1360,16 +1363,16 @@ class AsyncRequestUtils:
) )
if transport is not None: if transport is not None:
client = await stack.enter_async_context( client = await stack.enter_async_context(
httpx.AsyncClient( httpx2.AsyncClient(
transport=_NonClosingTransportProxy(transport), transport=_NonClosingTransportProxy(transport),
timeout=httpx.Timeout(self._timeout), timeout=httpx2.Timeout(self._timeout),
follow_redirects=self._follow_redirects, follow_redirects=self._follow_redirects,
cookies=cookies_dict, cookies=cookies_dict,
) )
) )
else: else:
client = await stack.enter_async_context( client = await stack.enter_async_context(
httpx.AsyncClient( httpx2.AsyncClient(
http2=self._http2, http2=self._http2,
proxy=self._proxies, proxy=self._proxies,
timeout=self._timeout, timeout=self._timeout,
@@ -1383,17 +1386,17 @@ class AsyncRequestUtils:
response = await stack.enter_async_context( response = await stack.enter_async_context(
client.stream("GET", url, params=params, **kwargs) client.stream("GET", url, params=params, **kwargs)
) )
except stale_conn_errs: except _ASYNC_STALE_CONNECTION_ERRORS:
try: try:
response = await stack.enter_async_context( response = await stack.enter_async_context(
client.stream("GET", url, params=params, **kwargs) client.stream("GET", url, params=params, **kwargs)
) )
except httpx.RequestError: except httpx2.RequestError:
if raise_exception: if raise_exception:
raise raise
yield None yield None
return return
except httpx.RequestError: except httpx2.RequestError:
if raise_exception: if raise_exception:
raise raise
yield None yield None
@@ -1413,7 +1416,7 @@ class AsyncRequestUtils:
json: dict = None, json: dict = None,
raise_exception: bool = False, raise_exception: bool = False,
**kwargs, **kwargs,
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
发送异步POST请求并返回响应对象 发送异步POST请求并返回响应对象
:param url: 请求的URL :param url: 请求的URL
@@ -1425,7 +1428,7 @@ class AsyncRequestUtils:
:param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None :param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None
:param kwargs: 其他请求参数,如headers, cookies, proxies等 :param kwargs: 其他请求参数,如headers, cookies, proxies等
:return: HTTP响应对象,若发生RequestError则返回None :return: HTTP响应对象,若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exceptionTrue时抛出 :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出
""" """
return await self.request( return await self.request(
method="post", method="post",
@@ -1449,7 +1452,7 @@ class AsyncRequestUtils:
json: dict = None, json: dict = None,
raise_exception: bool = False, raise_exception: bool = False,
**kwargs, **kwargs,
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
发送异步PUT请求并返回响应对象 发送异步PUT请求并返回响应对象
:param url: 请求的URL :param url: 请求的URL
@@ -1461,7 +1464,7 @@ class AsyncRequestUtils:
:param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None :param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None
:param kwargs: 其他请求参数,如headers, cookies, proxies等 :param kwargs: 其他请求参数,如headers, cookies, proxies等
:return: HTTP响应对象,若发生RequestError则返回None :return: HTTP响应对象,若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exceptionTrue时抛出 :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出
""" """
return await self.request( return await self.request(
method="put", method="put",
@@ -1483,7 +1486,7 @@ class AsyncRequestUtils:
allow_redirects: bool = True, allow_redirects: bool = True,
raise_exception: bool = False, raise_exception: bool = False,
**kwargs, **kwargs,
) -> Optional[httpx.Response]: ) -> Optional[httpx2.Response]:
""" """
发送异步DELETE请求并返回响应对象 发送异步DELETE请求并返回响应对象
:param url: 请求的URL :param url: 请求的URL
@@ -1493,7 +1496,7 @@ class AsyncRequestUtils:
:param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None :param raise_exception: 是否在发生异常时抛出异常,否则默认拦截异常返回None
:param kwargs: 其他请求参数,如headers, cookies, proxies等 :param kwargs: 其他请求参数,如headers, cookies, proxies等
:return: HTTP响应对象,若发生RequestError则返回None :return: HTTP响应对象,若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exceptionTrue时抛出 :raises: HTTP 客户端请求异常仅在 raise_exception=True 时抛出
""" """
return await self.request( return await self.request(
method="delete", method="delete",
+4 -2
View File
@@ -8,7 +8,7 @@ from random import choice
from typing import Optional, Union from typing import Optional, Union
from urllib import parse from urllib import parse
import httpx import httpx2
import requests import requests
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
@@ -223,7 +223,9 @@ class DoubanApi(metaclass=WeakSingleton):
return req_url, params return req_url, params
@staticmethod @staticmethod
def _handle_response(resp: Union[requests.Response, httpx.Response]) -> dict: def _handle_response(
resp: Union[requests.Response, httpx2.Response]
) -> dict:
""" """
处理HTTP响应 处理HTTP响应
""" """
+11
View File
@@ -122,6 +122,17 @@ dependencies = ["example-package>=1,<2"]
- 仅有 `requirements.txt` 的历史插件继续按原方式安装; - 仅有 `requirements.txt` 的历史插件继续按原方式安装;
- 宿主不消费插件自己的 `uv.lock`,因为多个插件共享同一主程序环境,不能分别同步独立锁文件。 - 宿主不消费插件自己的 `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. 准备资源与插件目录 ### 4. 准备资源与插件目录
本地源码开发时,主程序需要读取资源文件和插件源码。相关文件需要放到主程序实际加载的目录下: 本地源码开发时,主程序需要读取资源文件和插件源码。相关文件需要放到主程序实际加载的目录下:
+1
View File
@@ -33,6 +33,7 @@ dependencies = [
"fastapi~=0.141.1", "fastapi~=0.141.1",
"google-genai~=2.8.0", "google-genai~=2.8.0",
"httpx[http2,socks]~=0.28.1", "httpx[http2,socks]~=0.28.1",
"httpx2[http2,socks]~=2.12.0",
"jieba-next~=1.0.0rc1", "jieba-next~=1.0.0rc1",
"jinja2~=3.1.6", "jinja2~=3.1.6",
"langchain~=1.3.15", "langchain~=1.3.15",
+73 -5
View File
@@ -1,11 +1,12 @@
import asyncio import asyncio
import time import time
import httpx import httpx2
import pytest import pytest
from app.adapters.network import http as http_module from app.adapters.network import http as http_module
from app.adapters.network.http import AsyncRequestUtils from app.adapters.network.http import AsyncRequestUtils
from app.sdk.network import AsyncRequestUtils as SdkAsyncRequestUtils
PROXY = "http://proxy.example:7890" PROXY = "http://proxy.example:7890"
URL = "https://raw.githubusercontent.com/demo/repo/main/package.json" URL = "https://raw.githubusercontent.com/demo/repo/main/package.json"
@@ -24,7 +25,7 @@ def _fake_dispatch(calls, fail_when):
calls.append(http2) calls.append(http2)
if fail_when(http2): if fail_when(http2):
if raise_exception: if raise_exception:
raise httpx.RemoteProtocolError("tunnel closed") raise httpx2.RemoteProtocolError("tunnel closed")
return None return None
return "ok" 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): async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs):
calls.append(http2) calls.append(http2)
raise httpx.ConnectTimeout("proxy slow") raise httpx2.ConnectTimeout("proxy slow")
monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake) 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): async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs):
calls.append(http2) calls.append(http2)
raise httpx.ConnectTimeout("proxy slow") raise httpx2.ConnectTimeout("proxy slow")
monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake) monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake)
utils = AsyncRequestUtils(proxies={"https": PROXY}) utils = AsyncRequestUtils(proxies={"https": PROXY})
with pytest.raises(httpx.ConnectTimeout): with pytest.raises(httpx2.ConnectTimeout):
asyncio.run(utils.request("get", URL, raise_exception=True)) asyncio.run(utils.request("get", URL, raise_exception=True))
assert calls == [True] assert calls == [True]
@@ -154,3 +155,70 @@ def test_no_proxy_configured_skips_breaker_logic(monkeypatch):
assert result == "ok" assert result == "ok"
assert calls == [True] 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)
+4 -3
View File
@@ -6,6 +6,7 @@ from types import SimpleNamespace
from unittest.mock import MagicMock from unittest.mock import MagicMock
import httpx import httpx
import httpx2
import pytest import pytest
from starlette.applications import Starlette from starlette.applications import Starlette
from starlette.responses import JSONResponse, StreamingResponse from starlette.responses import JSONResponse, StreamingResponse
@@ -162,12 +163,12 @@ async def test_async_external_request_preserves_explicit_header() -> None:
"""异步外呼默认传播当前 ID,但不得覆盖调用方显式 trace 边界。""" """异步外呼默认传播当前 ID,但不得覆盖调用方显式 trace 边界。"""
observed = [] observed = []
async def respond(request: httpx.Request) -> httpx.Response: async def respond(request: httpx2.Request) -> httpx2.Response:
"""记录 MockTransport 收到的请求头。""" """记录 MockTransport 收到的请求头。"""
observed.append(request.headers[CORRELATION_ID_HEADER]) 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) utils = AsyncRequestUtils(client=client)
with correlation_scope("context-request"): with correlation_scope("context-request"):
await utils.request("GET", "https://example.com/default") await utils.request("GET", "https://example.com/default")
+1 -1
View File
@@ -658,7 +658,7 @@ def test_shared_http_close_waits_for_real_lru_eviction(monkeypatch):
raise RuntimeError("eviction close failed") raise RuntimeError("eviction close failed")
monkeypatch.setattr(http_utils, "_MAX_SHARED_TRANSPORTS_PER_LOOP", 1) 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(): async def run_test():
transport_kwargs = { transport_kwargs = {
"proxy": None, "proxy": None,
Generated
+47
View File
@@ -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" }, { 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]] [[package]]
name = "httpx" name = "httpx"
version = "0.28.1" 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" }, { 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]] [[package]]
name = "hyperframe" name = "hyperframe"
version = "6.1.0" version = "6.1.0"
@@ -1500,6 +1536,7 @@ dependencies = [
{ name = "fastapi" }, { name = "fastapi" },
{ name = "google-genai" }, { name = "google-genai" },
{ name = "httpx", extra = ["http2", "socks"] }, { name = "httpx", extra = ["http2", "socks"] },
{ name = "httpx2", extra = ["http2", "socks"] },
{ name = "jieba-next" }, { name = "jieba-next" },
{ name = "jinja2" }, { name = "jinja2" },
{ name = "langchain" }, { name = "langchain" },
@@ -1609,6 +1646,7 @@ requires-dist = [
{ name = "fastapi", specifier = "~=0.141.1" }, { name = "fastapi", specifier = "~=0.141.1" },
{ name = "google-genai", specifier = "~=2.8.0" }, { name = "google-genai", specifier = "~=2.8.0" },
{ name = "httpx", extras = ["http2", "socks"], specifier = "~=0.28.1" }, { 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 = "jieba-next", specifier = "~=1.0.0rc1" },
{ name = "jinja2", specifier = "~=3.1.6" }, { name = "jinja2", specifier = "~=3.1.6" },
{ name = "langchain", specifier = "~=1.3.15" }, { 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" }, { 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]] [[package]]
name = "typing-extensions" name = "typing-extensions"
version = "4.16.0" version = "4.16.0"