Skip to main content

🎯 可移植、可拔插的异步HTTP客户端 — httpx为主,Playwright自动降级,支持多代理模式与策略

Project description

🚀 xqcrequests

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

Python httpx 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_ONLYAUTO_FALLBACKBROWSER_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

流程要点:

  1. 主力 HttpxTransport 发送请求
  2. 若成功 → 直接返回 Response
  3. 若失败 → 4 个检测器并行判断是否需要降级
  4. 任一检测器命中 → 按顺序执行 FallbackChain 中的 Handler
  5. 首个成功返回的 Handler 结果即为最终响应
  6. 全部 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


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

xqcrequests-0.0.1.tar.gz (29.5 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

xqcrequests-0.0.1-py3-none-any.whl (30.1 kB view details)

Uploaded Python 3

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

Hashes for xqcrequests-0.0.1.tar.gz
Algorithm Hash digest
SHA256 95d3287035ee39f006da9715d31ead782925e563d6a9eeb5094be03746189380
MD5 4eeb6505e645976e50c2d14782f0a961
BLAKE2b-256 7a30468e541a94ee51242f73925c93470c005392de0752e39ee91f571fb0783b

See more details on using hashes here.

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

Hashes for xqcrequests-0.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 89c0cce5b0125ee23380b7f87d63f83556d3990b32d58938ff91838aeb9346ca
MD5 20d3218dedd53d18c23c11bf4ab00f65
BLAKE2b-256 1cf81b4b1bf8ba924ebbf43801bbe426ba82944e8c3a0dabab18618e7b21d9a7

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page