mipu-open/sdk/mipu_requests/MIGRATION.md

mipu_requests — 从 requests 迁移指南

快速对比

# ── requests ──────────────────────────────────────
import requests

s = requests.Session()
s.proxies = {"https": "http://proxy:8080"}
resp = s.get("https://example.com/page")
resp.raise_for_status()
data = resp.json()

# ── mipu_requests ─────────────────────────────────
import mipu_requests

s = mipu_requests.session(
    base_url="https://api-eu.mipuyun.com",
    client_key="your_client_key",
    client_secret="your_client_secret",
    agent="JQ",
)
s.proxies = "res_mipu_bookxWdyg"
resp = s.get("https://example.com/page")
resp.raise_for_status()
data = resp.json()

安装

pip install mipu-requests

创建 Session

import mipu_requests

# Agent 模式 — 通过指定 agent 走 /unlocker/agent 端点
s = mipu_requests.session(
    base_url="https://api-eu.mipuyun.com",
    client_key="your_key",
    client_secret="your_secret",
    agent="JQ",
)

# Request 模式 — 走 /unlocker/request 端点（无需 agent）
s = mipu_requests.session(
    base_url="https://api-eu.mipuyun.com",
    client_key="your_key",
    client_secret="your_secret",
    mode="request",
)

设置代理

与 requests 唯一必须修改的地方：

# requests — 传 dict
s.proxies = {"http": "http://proxy:8080", "https": "http://proxy:8080"}

# mipu_requests — 传字符串（米普 proxy 标识）
s.proxies = "res_mipu_bookxWdyg"

# 单次请求覆盖代理
resp = s.get(url, proxy="another_proxy")

请求方法

完全一致，无需改动：

resp = s.get(url, params={"key": "value"}, headers={...})
resp = s.post(url, data={"key": "value"}, headers={...})
resp = s.post(url, json={"key": "value"}, headers={...})
resp = s.put(url, data=body, headers={...})
resp = s.delete(url, headers={...})

Response 对象

属性和方法与 requests.Response 一致：

resp.status_code     # int   HTTP 状态码
resp.text            # str   响应体文本
resp.content         # bytes 响应体字节
resp.headers         # dict  响应头
resp.ok              # bool  status_code 在 200-399
resp.url             # str   请求 URL
resp.encoding        # str   编码（默认 utf-8）
resp.cookies         # dict  空 dict（兼容属性）
resp.reason          # str   状态码原因短语

resp.json()                # 解析 JSON
resp.raise_for_status()    # >=400 抛异常
resp.iter_content(chunk_size)  # 按块迭代

异常处理

try:
    resp = s.get(url)
    resp.raise_for_status()
except mipu_requests.HTTPError as e:
    print(f"目标站错误: {e.response.status_code}")
except mipu_requests.APIError as e:
    print(f"API 错误: {e}")
except mipu_requests.RequestException as e:
    print(f"请求异常: {e}")

异步模式

默认开启异步轮询防止长连接断连：

# 默认异步
s = mipu_requests.session(...)

# 关闭异步
s = mipu_requests.session(..., async_mode=False)

# 自定义参数
s = mipu_requests.session(
    ...,
    poll_interval=3,
    poll_max=20,
    submit_timeout=60,
    poll_timeout=(5, 30),
)

完整迁移示例

# ── 迁移前 (requests) ────────────────────────────
import requests

s = requests.Session()
s.headers.update({"user-agent": "Mozilla/5.0"})
s.proxies = {"https": "http://proxy:8080"}

resp = s.get("https://example.com/api", params={"page": 1})
resp.raise_for_status()
data = resp.json()

# ── 迁移后 (mipu_requests) ───────────────────────
import mipu_requests

s = mipu_requests.session(
    base_url="https://api-eu.mipuyun.com",
    client_key="your_key",
    client_secret="your_secret",
    agent="JQ",
)
s.headers = {"user-agent": "Mozilla/5.0"}
s.proxies = "res_mipu_bookxWdyg"

resp = s.get("https://example.com/api", params={"page": 1})
resp.raise_for_status()
data = resp.json()

迁移只需 2 步：

1. 构造 mipu_requests.session() 时传入 base_url/client_key/client_secret/agent
2. s.proxies = {...} → s.proxies = "proxy_name"

其余代码无需改动。