多引擎切换

hs-net 支持 5 种 HTTP 引擎,切换只需改一个参数。

章节概览

章节说明示例文件
引擎一览5 种引擎对比multi_engine.py
切换引擎字符串/枚举指定引擎multi_engine.py
引擎特定配置engine_options 透传multi_engine.py
自定义引擎继承 EngineBase

引擎一览

引擎异步同步HTTP/2TLS 指纹SOCKS 代理安装方式适用场景
httpx默认通用场景(默认)
aiohttp[aiohttp]高并发异步
curl-cffi[curl]反爬、爬虫
requests[requests]简单同步脚本
requests-go[requests-go]追求性能

切换引擎

字符串方式

engine_string.py
# 异步
async with Net(engine="aiohttp") as net:
    resp = await net.get("https://example.com")

# 同步
with SyncNet(engine="requests") as net:
    resp = net.get("https://example.com")

枚举方式

engine_enum.py
from hs_net import EngineEnum

with SyncNet(engine=EngineEnum.CURL_CFFI) as net:
    resp = net.get("https://example.com")

引擎特定配置

通过 engine_options 传递引擎特有参数:

httpx

httpx_options.py
# 关闭 HTTP/2(默认开启)
net = SyncNet(engine="httpx", engine_options={"http2": False})

curl-cffi

curl_cffi_options.py
# 模拟 Chrome 120 的 TLS 指纹
net = SyncNet(
    engine="curl_cffi",
    engine_options={
        "impersonate": "chrome120",   # 浏览器指纹
    },
)
支持的指纹

curl-cffi 支持多种浏览器指纹:chrome110chrome120safari15_5edge101 等。 详见 curl-cffi 文档

不兼容的引擎

# ❌ requests 不支持异步
Net(engine="requests")  # 抛出 ValueError

# ❌ aiohttp 不支持同步
SyncNet(engine="aiohttp")  # 抛出 ValueError

引擎未安装

使用未安装的引擎时,会抛出 EngineNotInstalled 异常并提示安装命令:

from hs_net import SyncNet, EngineNotInstalled

try:
    net = SyncNet(engine="curl_cffi")
except EngineNotInstalled as e:
    print(e)
    # [EngineNotInstalled]: 引擎 curl-cffi 需要额外安装依赖: pip install hs-net[curl]

自定义引擎

你可以继承 EngineBase(异步)或 SyncEngineBase(同步)实现自定义引擎:

custom_engine.py
from hs_net.engines.base import SyncEngineBase
from hs_net.models import RequestModel
from hs_net.response import Response

class MySyncEngine(SyncEngineBase):
    def __init__(self, sem=None, headers=None, cookies=None, verify=True, **engine_options):
        super().__init__(sem, headers, cookies, verify, **engine_options)
        # 初始化你的 HTTP 客户端

    def close(self):
        # 关闭客户端

    @property
    def cookies(self) -> dict[str, str]:
        return {}

    def _download(self, request_data: RequestModel) -> Response:
        # 实现请求逻辑
        ...

# 使用自定义引擎
with SyncNet(engine=MySyncEngine) as net:
    resp = net.get("https://example.com")