🚀 xqcrequests

可移植 · 可拔插 · 自动降级的异步 HTTP 客户端

Python Playwright License

✨ 特性

🎯 核心能力	说明
⚡ 双引擎	httpx 高性能请求 + Playwright 浏览器渲染，自动切换
🔄 智能降级	检测反爬 / 超时 / 服务端错误 / 内容异常，自动触发降级
🌐 多代理模式	直连 / 单代理 / 代理池 / API 接口，一键切换
🎲 4种策略	轮询 / 随机(失败剔除) / 粘性(优选上次成功) / 加权
🔌 完全可拔插	传输层 / 代理源 / 检测器 / 降级处理器均可自定义替换
📦 零项目依赖	独立包，可直接移植到任何 Python 项目
🛠️ 灵活配置	config 对象 / 直接参数 / 混合使用，三种方式任选

📦 安装

pip install xqcrequests
# 或
poetry add xqcrequests

依赖

python >= 3.10
httpx >= 0.25.0
playwright >= 1.40.0

Playwright 浏览器安装

Playwright 需要安装浏览器才能使用浏览器模式（BROWSER_ONLY、AUTO_FALLBACK、BROWSER_FIRST）。

安装 Playwright 后，必须运行以下命令安装浏览器：

# 安装 chromium 浏览器
python -m playwright install chromium

# 安装所有浏览器（chromium、firefox、webkit）
python -m playwright install

# 安装浏览器及系统依赖（推荐，Linux/macOS 需要）
python -m playwright install --with-deps

代码中检测浏览器环境：

from xqcrequests import check_browser_environment, ensure_browser_available

# 检测浏览器环境状态
status = check_browser_environment()
print(f"Chromium: {'已安装' if status['chromium_installed'] else '未安装'}")

# 确保浏览器可用，不可用时显示安装提示
if ensure_browser_available():
    # 可以安全使用浏览器模式
    pass

Docker / CI 环境安装示例：

# 安装浏览器（不安装系统依赖）
RUN pip install playwright && python -m playwright install chromium

# 或安装所有浏览器
RUN pip install playwright && python -m playwright install

🏗️ 架构概览

设计目标

目标	实现方式
可移植	零项目耦合，仅依赖 `httpx` + `playwright`，可复制到任何项目直接使用
可拔插	每层（传输/代理/降级）均基于 ABC 抽象基类，用户可自由替换实现
自动降级	httpx 主请求失败时，自动检测反爬/网络/服务端异常并降级到 Playwright
统一入口	`HttpClient` 一个类覆盖所有场景，用户无需关心底层细节

分层架构

┌─────────────────────────────────────────────────────┐
│                    用户代码                          │
│          async with HttpClient() as client:          │
│              resp = await client.get(url)            │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│              HttpClient (统一入口)                   │
│  ┌─────────┬──────────┬──────────┬────────────────┐  │
│  │ 配置解析 │ 模式路由  │ 代理获取  │  降级链管理     │  │
│  └────┬────┴─────┬────┴─────┬────┴────────┬───────┘  │
│       │         │         │             │           │
│  ┌────▼────┐ ┌──▼────┐ ┌──▼──────┐ ┌────▼────────┐ │
│  │HttpxTrans│BrowserT │ProxyProv.│FallbackChain  │ │
│  │port     │ransport │ider      │              │ │
│  └─────────┘└────────┘└─────────┘└──────────────┘ │
└─────────────────────────────────────────────────────┘

三层职责

层级	目录	职责	可替换点
🚂 传输层	`transport/`	发送 HTTP 请求，返回 Response	继承 `BaseTransport`（如替换为 aiohttp）
🌐 代理层	`proxy/`	管理代理获取与选择策略	继承 `BaseProxyProvider` + `BaseProxyStrategy`
🔄 降级层	`fallback/`	检测异常 → 触发降级 → 链式处理	继承 `BaseFallbackHandler` + 自定义检测器

🚀 快速开始

最简用法 — 一行代码

import asyncio
from xqcrequests import HttpClient

async def main():
    async with HttpClient() as client:
        resp = await client.get("https://httpbin.org/get")
        print(resp.status_code)   # 200
        print(resp.ok)            # True
        print(resp.text[:100])    # 响应内容

asyncio.run(main())

带代理 + 自定义超时

from xqcrequests import HttpClient

async with HttpClient(
    proxy="http://127.0.0.1:7890",
    timeout=15.0,
    max_retries=3,
) as client:
    resp = await client.get("https://example.com")

使用完整配置对象

from xqcrequests import (
    HttpClient,
    HttpClientConfig,
    ProxyClientConfig,
    ProxyMode,
    ProxyStrategy,
    RequestMode,
    TimeoutConfig,
)

config = HttpClientConfig(
    mode=RequestMode.AUTO_FALLBACK,
    timeout=TimeoutConfig(connect=5.0, read=20.0),
    proxy=ProxyClientConfig(
        mode=ProxyMode.POOL,
        pool_urls=["http://p1:8080", "http://p2:8080"],
        strategy=ProxyStrategy.RANDOM,
    ),
)

async with HttpClient(config=config) as client:
    resp = await client.post("https://api.example.com/data", json={"key": "value"})

🎮 4种请求模式

模式	枚举值	行为说明	适用场景
HTTPX_ONLY	`"httpx_only"`	仅使用 httpx，不启用降级	API 接口、JSON 数据
BROWSER_ONLY	`"browser_only"`	仅使用 Playwright 浏览器	强反爬页面、JS 渲染
⭐ AUTO_FALLBACK	`"auto_fallback"`	默认模式。httpx 优先，失败时自动降级到浏览器	通用场景，兼顾性能与可靠性
BROWSER_FIRST	`"browser_first"`	浏览器优先，失败时降级到 httpx	高度反爬目标

client = HttpClient(mode="auto_fallback")       # 字符串
client = HttpClient(mode=RequestMode.BROWSER_FIRST)  # 枚举

🔄 请求流程详解（AUTO_FALLBACK 模式）

用户调用 client.get(url)
        │
        ▼
  ┌──────────────┐
  │ 获取代理(可选) │
  └──────┬───────┘
         │
         ▼
  ┌──────────────┐
  │ HttpxTransport │ ◄── 主传输层
  │  发送请求      │
  └──────┬───────┘
         │
    ┌────┴────┐
    │ 成功？   │
    └────┬────┘
     是 │ │ 否
        │ ▼
  ┌─────────────┐
  │ 4种检测器扫描  │ ◄── 反爬/网络错误/5xx/内容异常
  └──────┬──────┘
         │ 触发降级
         ▼
  ┌─────────────────┐
  │ FallbackChain    │
  │  ├─ browser_fallback (内置)
  │  ├─ 用户自定义 handler 1
  │  └─ 用户自定义 handler 2
  └────────┬────────┘
           │
           ▼
       返回 Response

流程要点：

主力 HttpxTransport 发送请求
若成功 → 直接返回 Response
若失败 → 4 个检测器并行判断是否需要降级
任一检测器命中 → 按顺序执行 FallbackChain 中的 Handler
首个成功返回的 Handler 结果即为最终响应
全部 Handler 失败 → 抛出 FallbackExhaustedError

🌐 代理系统

4种代理模式

# ① 直连 — 无代理（默认）
HttpClient(proxy=None)

# ② 单代理 — 固定地址
HttpClient(proxy="http://127.0.0.1:7890")

# ③ 代理池 — 多个地址轮换
HttpClient(config=HttpClientConfig(
    proxy=ProxyClientConfig(
        mode=ProxyMode.POOL,
        pool_urls=["http://p1:8080", "http://p2:8080", "http://p3:8080"],
        strategy=ProxyStrategy.ROUND_ROBIN,
    )
))

# ④ API 接口 — 动态获取代理列表
HttpClient(config=HttpClientConfig(
    proxy=ProxyClientConfig(
        mode=ProxyMode.API,
        api=ProxyAPIConfig(
            url="https://proxy-api.example.com/proxies",
            key="your-api-key",
            interval=120,
        ),
        strategy=ProxyStrategy.WEIGHTED,
    )
))

4种选择策略

策略	行为	场景
`ROUND_ROBIN`	依次轮换，均衡使用	通用负载均衡
`RANDOM`	随机选择，失败自动移除	快速淘汰坏节点
`STICKY`	优先使用上次成功的代理	保持会话一致性
`WEIGHTED`	按权重随机，成功增权/失败减权	自适应质量排序

🔄 自动降级机制

内置检测器

当主传输层请求失败时，以下检测器判断是否触发降级：

检测器	名称	触发条件
🛡️ 反爬检测	`anti_scrap`	响应含验证码/captcha/security verify 等关键词
🌐 网络异常	`network`	超时 / 连接拒绝 / DNS 失败等网络层异常
🖥️ 服务端错误	`server_error`	HTTP 500/502/503/504
📄 内容异常	`content_anomaly`	响应为空或过短 (<100 字节)

降级链执行逻辑

主请求失败
    │
    ▼
遍历所有启用的检测器 ──── 匹配到任一检测器？
    │                         │
    │ 是                      │ 否（但仍记录警告日志）
    │                         │
    ▼                         │
按顺序执行 FallbackHandler 链  │
    │                         │
    ├─ handler1.handle() ──→ 返回 Response? ──→ ✅ 成功返回
    │         │ 否            │
    │         ▼               │
    ├─ handler2.handle() ──→ 返回 Response? ──→ ✅ 成功返回
    │         │ 否            │
    │         ▼               │
    ├─ handlerN.handle() ──→ 返回 Response? ──→ ✅ 成功返回
    │         │ 否            │
    │         ▼               │
    └─ 全部失败 → ❌ FallbackExhaustedError

自定义降级处理器

from xqcrequests import HttpClient, BaseFallbackHandler, Response, FallbackConfig

class MyCacheHandler(BaseFallbackHandler):
    """缓存兜底处理器"""

    @property
    def name(self) -> str:
        return "cache_fallback"

    async def handle(self, method: str, url: str, error: Exception, **kwargs) -> Response | None:
        cached = self._cache.get(url)
        if cached:
            return Response(status_code=200, text=cached, url=url)
        return None  # None = 不处理，交给下一个

# 方式1: 通过配置注册
handler = MyCacheHandler()
client = HttpClient(config=HttpClientConfig(
    fallback=FallbackConfig(custom_handlers=[handler]),
))

# 方式2: 运行时动态管理
client.add_fallback_handler(handler)
client.remove_fallback_handler("cache_fallback")

📡 Response 对象

所有请求方法返回统一的 Response 对象：

resp = await client.get("https://example.com")

resp.status_code          # int   — HTTP 状态码
resp.headers              # dict  — 响应头
resp.text                 # str   — 文本内容 (UTF-8)
resp.content              # bytes — 二进制内容
resp.url                  # str   — 最终 URL（跟随重定向后）
resp.encoding             # str   — 编码（默认 utf-8）
resp.ok                   # bool  — status_code < 400?

resp.json()               # dict  — 解析 JSON
resp.raise_for_status()          # 非 2xx 抛 TransportError
repr(resp)              # "Response(status_code=200, url='...', len=1234)"

🔧 配置体系

支持 3 种配置方式，可混合使用：

# 方式1: HttpClientConfig 完整配置对象
config = HttpClientConfig(mode=RequestMode.AUTO_FALLBACK, ...)
client = HttpClient(config=config)

# 方式2: 直接传参（快捷方式）
client = HttpClient(
    mode="auto_fallback",
    proxy="http://proxy:8080",
    timeout=15.0,
    max_retries=3,
)

# 方式3: 混合使用（config + 参数覆盖）
base_config = HttpClientConfig()
client = HttpClient(
    config=base_config,
    mode="browser_only",       # 覆盖 base_config 的 mode
    timeout=5.0,               # 覆盖超时
)

HttpClient 构造参数速查

参数	类型	默认值	说明
`config`	`HttpClientConfig \| None`	`None`	完整配置对象
`mode`	`str \| RequestMode \| None`	`None`	请求模式（覆盖 config）
`timeout`	`float \| TimeoutConfig \| None`	`None`	超时设置（数字=统一超时）
`proxy`	`str \| ProxyClientConfig \| None`	`None`	代理配置（字符串=单代理 URL）
`headers`	`dict \| None`	`None`	默认请求头
`follow_redirects`	`bool \| None`	`None`	是否跟随重定向
`verify_ssl`	`bool \| None`	`None`	是否验证 SSL 证书
`headless`	`bool \| None`	`None`	浏览器无头模式
`user_agent`	`str \| None`	`None`	User-Agent 字符串
`max_retries`	`int \| RetryConfig \| None`	`None`	最大重试次数
`**kwargs`	—	—	直接覆盖 config 属性

HttpClientConfig 数据类字段

字段	类型	默认值	说明
`mode`	`RequestMode`	`AUTO_FALLBACK`	请求模式
`timeout`	`TimeoutConfig`	`connect=10, read=30`	超时配置
`retry`	`RetryConfig`	`max_attempts=3`	重试策略
`concurrency`	`ConcurrencyConfig`	`max_connections=100`	并发控制
`browser`	`BrowserConfig`	`headless=True`	浏览器行为配置
`proxy`	`ProxyClientConfig`	`DIRECT`	代理配置
`fallback`	`FallbackConfig`	`enabled=True`	降级配置
`default_headers`	`dict`	`{}`	默认请求头
`follow_redirects`	`bool`	`True`	跟随重定向
`verify_ssl`	`bool`	`True`	SSL 验证

子配置对象

TimeoutConfig

字段	类型	默认值	说明
`connect`	`float`	`10.0`	连接超时 (秒)
`read`	`float`	`30.0`	读取超时 (秒)
`write`	`float`	`30.0`	写入超时 (秒)
`pool`	`float`	`30.0`	连接池获取超时 (秒)

RetryConfig

字段	类型	默认值	说明
`max_attempts`	`int`	`3`	最大重试次数
`base_delay`	`float`	`1.0`	初始延迟 (秒)
`max_delay`	`float`	`30.0`	最大延迟 (秒)
`exponential_base`	`float`	`2.0`	指数退避基数

ProxyClientConfig

字段	类型	默认值	说明
`mode`	`ProxyMode`	`DIRECT`	代理模式
`single_url`	`str`	`""`	单代理 URL（SINGLE 模式）
`pool_urls`	`list[str]`	`[]`	代理池列表（POOL 模式）
`strategy`	`ProxyStrategy`	`ROUND_ROBIN`	选择策略
`api`	`ProxyAPIConfig`	`-`	API 代理源配置

BrowserConfig

字段	类型	默认值	说明
`headless`	`bool`	`True`	无头模式
`timeout`	`int`	`30000`	页面加载超时 (ms)
`user_agent`	`str`	Chrome UA	浏览器 UA
`viewport_width`	`int`	`1920`	视口宽度
`viewport_height`	`int`	`1080`	视口高度
`locale`	`str`	`"zh-CN"`	语言区域

FallbackConfig

字段	类型	默认值	说明
`enabled`	`bool`	`True`	是否启用降级
`detectors`	`list[str]`	全部 4 个	启用的检测器名称列表
`custom_handlers`	`list`	`[]`	用户自定义处理器列表

🧩 可拔插扩展

自定义传输层

继承 BaseTransport 即可实现自定义传输后端：

from xqcrequests import BaseTransport, Response

class AiohttpTransport(BaseTransport):
    """使用 aiohttp 替代 httpx"""

    def __init__(self, timeout: float = 30):
        self._timeout = timeout
        self._session = None

    async def _ensure_session(self):
        if self._session is None:
            import aiohttp
            self._session = aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=self._timeout)
            )

    async def get(self, url: str, **kwargs) -> Response:
        await self._ensure_session()
        async with self._session.get(url, **kwargs) as resp:
            return Response(
                status_code=resp.status,
                headers=dict(resp.headers),
                text=await resp.text(),
                url=str(resp.url),
            )

    async def post(self, url: str, **kwargs) -> Response:
        await self._ensure_session()
        async with self._session.post(url, **kwargs) as resp:
            return Response(
                status_code=resp.status,
                headers=dict(resp.headers),
                text=await resp.text(),
                url=str(resp.url),
            )
    # 同理实现 put/delete/head/patch/options/close

自定义代理提供者

from xqcrequests import BaseProxyProvider, ProxyInfo

class RedisProxyProvider(BaseProxyProvider):
    """从 Redis 获取代理"""

    async def acquire(self) -> ProxyInfo | None:
        url = await self._redis.spop("proxies")
        return ProxyInfo(url=url) if url else None

    async def release(self, proxy: ProxyInfo, success: bool) -> None:
        if success:
            await self._redis.sadd("proxies", proxy.url)

自定义检测器

from xqcrequests.fallback.detectors import DetectionResult

class RateLimitDetector:
    name = "rate_limit"

    def detect(self, response, error) -> DetectionResult:
        if response and response.status_code == 429:
            return DetectionResult(True, "触发速率限制", self.name)
        return DetectionResult(False)

📘 完整使用示例

示例 1：最简用法

import asyncio
from xqcrequests import HttpClient

async def main():
    async with HttpClient() as client:
        resp = await client.get("https://httpbin.org/get")
        print(resp.status_code)  # 200
        print(resp.json())  # {"origin": "..."}

asyncio.run(main())

示例 2：带代理的爬虫

from xqcrequests import (
    HttpClient, ProxyClientConfig, ProxyMode, ProxyStrategy,
    RequestMode, TimeoutConfig,
)

async def scrape_with_proxy():
    async with HttpClient(
        mode=RequestMode.AUTO_FALLBACK,
        timeout=TimeoutConfig(connect=5, read=60),
        proxy=ProxyClientConfig(
            mode=ProxyMode.POOL,
            pool_urls=[
                "http://proxy1.example.com:8080",
                "http://proxy2.example.com:8080",
                "http://proxy3.example.com:8080",
            ],
            strategy=ProxyStrategy.STICKY,
        ),
        headers={
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                          "AppleWebKit/537.36 Chrome/120.0.0.0",
            "Accept": "text/html,application/xhtml+xml",
        },
    ) as client:
        resp = await client.get("https://target-website.com/data")
        print(resp.text[:500])

示例 3：强反爬站点（纯浏览器模式）

from xqcrequests import HttpClient, RequestMode

async def scrape_anti_bot_site():
    async with HttpClient(
        mode=RequestMode.BROWSER_ONLY,
        headless=False,  # 有头模式，方便调试
    ) as client:
        resp = await client.get("https://cloudflare-protected-site.com")
        print(resp.text)

示例 4：POST 请求 + JSON

from xqcrequests import HttpClient

async def post_json():
    async with HttpClient() as client:
        resp = await client.post(
            "https://httpbin.org/post",
            json={"key": "value", "nested": {"a": 1}},
        )
        data = resp.json()
        print(data["json"])

示例 5：自定义重试 + 降级配置

from xqcrequests import (
    HttpClient, HttpClientConfig,
    RetryConfig, TimeoutConfig,
    FallbackConfig, ProxyClientConfig, ProxyMode, RequestMode,
)

async def robust_request():
    config = HttpClientConfig(
        mode=RequestMode.AUTO_FALLBACK,
        timeout=TimeoutConfig(connect=5, read=30, write=10),
        retry=RetryConfig(max_attempts=5, base_delay=2.0, max_delay=60),
        proxy=ProxyClientConfig(mode=ProxyMode.DIRECT),
        fallback=FallbackConfig(
            enabled=True,
            detectors=["anti_scrap", "network", "server_error"],
        ),
    )
    async with HttpClient(config=config) as client:
        resp = await client.get("https://unstable-api.com/endpoint")
        return resp

🧪 测试

poetry install --with dev
pytest tests/ -v

文件	内容
`test_core.py`	Response / 配置 / 枚举 / 异常体系
`test_proxy.py`	Provider 工厂 / 4 种策略行为
`test_fallback.py`	4 种检测器 / FallbackChain / 自定义 Handler
`test_client.py`	HttpClient 实例化 / 4 种模式构建 / 动态管理
`test_examples.py`	8 个真实使用场景，从最简到完整配置

📁 项目结构

xqcrequests/
├── pyproject.toml                 # Poetry 包配置
├── README.md                      # 本文档
├── .gitignore                     # Git 忽略规则
├── pytest.ini                     # pytest 配置
│
├── src/
│   └── xqcrequests/              # 🔥 核心代码
│       ├── __init__.py            # 48+ 个公共符号导出
│       ├── client.py              # HttpClient 统一入口
│       ├── config.py              # 12 个配置数据类 + 3 枚举
│       ├── exceptions.py          # 11 个异常类
│       │
│       ├── transport/             # 🚂 传输层
│       │   ├── base.py            # Response + BaseTransport ABC
│       │   ├── httpx_transport.py # HttpxTransport 实现
│       │   └── browser_transport.py # BrowserTransport (Playwright)
│       │
│       ├── proxy/                 # 🌐 代理层
│       │   ├── base.py            # BaseProxyProvider + ProxyInfo
│       │   ├── providers.py       # 4 种 Provider + 工厂函数
│       │   └── strategies.py      # 4 种策略 + 工厂函数
│       │
│       └── fallback/              # 🔄 降级层
│           ├── base.py            # BaseFallbackHandler ABC
│           ├── chain.py           # FallbackChain 链式管理
│           └── detectors.py       # 4 种内置检测器 + 注册表
│
└── tests/
    ├── conftest.py                # src 布局路径配置
    ├── test_core.py               # 核心功能测试
    ├── test_proxy.py              # 代理系统测试
    ├── test_fallback.py           # 降级机制测试
    ├── test_client.py             # HttpClient 集成测试
    └── test_examples.py           # 📘 8 个真实使用场景示例

Made with ❤️ by Xiaoqiang

Release files for xqcrequests 0.0.1

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for xqcrequests 0.0.1
File	Size	Uploaded
xqcrequests-0.0.1.tar.gz	29.5 kB	Apr 23, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for xqcrequests 0.0.1
File	Interpreter	ABI	Platform	Reset
xqcrequests-0.0.1-py3-none-any.whl 30.1 kB Apr 23, 2026	Python 3	none	any	Details

Total release size: 59.6 kB

Release files / xqcrequests-0.0.1.tar.gz

Download URL	xqcrequests-0.0.1.tar.gz
Size	29.5 kB
Tags	Source
SHA-256 checksum How to use checksums	`95d3287035ee39f006da9715d31ead782925e563d6a9eeb5094be03746189380`
BLAKE2b-256 checksum How to use checksums	`7a30468e541a94ee51242f73925c93470c005392de0752e39ee91f571fb0783b`
Upload date	Apr 23, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/2.1.3 CPython/3.13.3 Windows/11`

Release files / xqcrequests-0.0.1-py3-none-any.whl

Download URL	xqcrequests-0.0.1-py3-none-any.whl
Size	30.1 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`89c0cce5b0125ee23380b7f87d63f83556d3990b32d58938ff91838aeb9346ca`
BLAKE2b-256 checksum How to use checksums	`1cf81b4b1bf8ba924ebbf43801bbe426ba82944e8c3a0dabab18618e7b21d9a7`
Upload date	Apr 23, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/2.1.3 CPython/3.13.3 Windows/11`

Release history Release notifications | RSS feed

This release

0.0.1 This release

Apr 23, 2026 2 release files

xqcrequests 0.0.1