🎯 可移植、可拔插的异步HTTP客户端 — httpx为主,Playwright自动降级,支持多代理模式与策略
Project description
🚀 xqcrequests
可移植 · 可拔插 · 自动降级的异步 HTTP 客户端
✨ 特性
| 🎯 核心能力 | 说明 |
|---|---|
| ⚡ 双引擎 | 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
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file xqcrequests-0.0.1.tar.gz.
File metadata
- Download URL: xqcrequests-0.0.1.tar.gz
- Upload date:
- Size: 29.5 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/2.1.3 CPython/3.13.3 Windows/11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
95d3287035ee39f006da9715d31ead782925e563d6a9eeb5094be03746189380
|
|
| MD5 |
4eeb6505e645976e50c2d14782f0a961
|
|
| BLAKE2b-256 |
7a30468e541a94ee51242f73925c93470c005392de0752e39ee91f571fb0783b
|
File details
Details for the file xqcrequests-0.0.1-py3-none-any.whl.
File metadata
- Download URL: xqcrequests-0.0.1-py3-none-any.whl
- Upload date:
- Size: 30.1 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/2.1.3 CPython/3.13.3 Windows/11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
89c0cce5b0125ee23380b7f87d63f83556d3990b32d58938ff91838aeb9346ca
|
|
| MD5 |
20d3218dedd53d18c23c11bf4ab00f65
|
|
| BLAKE2b-256 |
1cf81b4b1bf8ba924ebbf43801bbe426ba82944e8c3a0dabab18618e7b21d9a7
|